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

feat(velocity): migrate D2 sales source → objective_corpus_room_month (#156) #157

Closed
lekss361 wants to merge 4 commits from feat/migrate-velocity-to-objective-data into main
pull from: feat/migrate-velocity-to-objective-data
merge into: lekss361:main
Conversation 1 Commits 4 Files changed 6 +289 -265
6 changed files with 289 additions and 265 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

116
backend/app/api/v1/parcels.py
View file

@ -1658,33 +1658,34 @@ def analyze_parcel(
"lon": centroid_lon, "lon": centroid_lon,
} }
try: try:
zoning_row = ( with db.begin_nested():
db.execute( zoning_row = (
text(""" db.execute(
SELECT zone_code, zone_name, description, rosreestr_id text("""
FROM pzz_zones_ekb SELECT zone_code, zone_name, description, rosreestr_id
WHERE ST_Within( FROM pzz_zones_ekb
ST_Centroid(ST_GeomFromText(:wkt, 4326)), WHERE ST_Within(
geom ST_Centroid(ST_GeomFromText(:wkt, 4326)),
) geom
LIMIT 1 )
"""), LIMIT 1
{"wkt": geom_wkt}, """),
) {"wkt": geom_wkt},
.mappings() )
.first() .mappings()
) .first()
if zoning_row:
zoning.update(
{
"zone_code": zoning_row["zone_code"],
"zone_name": zoning_row["zone_name"],
"description": zoning_row["description"],
"rosreestr_id": zoning_row["rosreestr_id"],
"data_available": True,
"source": "rosreestr-pkk6-cached",
}
) )
if zoning_row:
zoning.update(
{
"zone_code": zoning_row["zone_code"],
"zone_name": zoning_row["zone_name"],
"description": zoning_row["description"],
"rosreestr_id": zoning_row["rosreestr_id"],
"data_available": True,
"source": "rosreestr-pkk6-cached",
}
)
except Exception as e: except Exception as e:
logger.warning("zoning query failed for %s: %s", cad_num, e) logger.warning("zoning query failed for %s: %s", cad_num, e)
@ -1692,9 +1693,10 @@ def analyze_parcel(
success_recommendation: dict[str, Any] | None = None success_recommendation: dict[str, Any] | None = None
if district_row: if district_row:
try: try:
success_rows = ( with db.begin_nested():
db.execute( success_rows = (
text(""" db.execute(
text("""
SELECT bucket, success_score, n_deals, avg_price_per_m2, avg_area_m2, SELECT bucket, success_score, n_deals, avg_price_per_m2, avg_area_m2,
velocity_z, price_z, area_z velocity_z, price_z, area_z
FROM v_bucket_success_score FROM v_bucket_success_score
@ -1702,35 +1704,35 @@ def analyze_parcel(
ORDER BY success_score DESC ORDER BY success_score DESC
LIMIT 5 LIMIT 5
"""), """),
{"dn": district_row["district_name"]}, {"dn": district_row["district_name"]},
)
.mappings()
.all()
) )
.mappings() if success_rows:
.all() success_recommendation = {
) "district": district_row["district_name"],
if success_rows: "ranking": [
success_recommendation = { {
"district": district_row["district_name"], "bucket": r["bucket"],
"ranking": [ "success_score": round(float(r["success_score"]), 2),
{ "n_deals": int(r["n_deals"]),
"bucket": r["bucket"], "avg_price_per_m2": (
"success_score": round(float(r["success_score"]), 2), int(r["avg_price_per_m2"]) if r["avg_price_per_m2"] else None
"n_deals": int(r["n_deals"]), ),
"avg_price_per_m2": ( "avg_area_m2": (
int(r["avg_price_per_m2"]) if r["avg_price_per_m2"] else None round(float(r["avg_area_m2"]), 1) if r["avg_area_m2"] else None
), ),
"avg_area_m2": ( }
round(float(r["avg_area_m2"]), 1) if r["avg_area_m2"] else None for r in success_rows
), ],
} "top_bucket": dict(success_rows[0]) if success_rows else None,
for r in success_rows "note": (
], "Топ комнатность по 'успешности' = z-scores: velocity×0.5 + price×0.3 "
"top_bucket": dict(success_rows[0]) if success_rows else None, "- area×0.2. Min 30 сделок в группе за 24 мес. "
"note": ( "Используй для квартирографии проекта."
"Топ комнатность по 'успешности' = z-scores: velocity×0.5 + price×0.3 " ),
"- area×0.2. Min 30 сделок в группе за 24 мес. " }
"Используй для квартирографии проекта."
),
}
except Exception as e: except Exception as e:
logger.warning("success_recommendation query failed for %s: %s", cad_num, e) logger.warning("success_recommendation query failed for %s: %s", cad_num, e)
success_recommendation = None success_recommendation = None

228
backend/app/services/analytics_queries.py
View file

@ -673,7 +673,14 @@ def object_detail(db: Session, obj_id: int) -> dict[str, Any] | None:
def object_sale_graph( def object_sale_graph(
db: Session, obj_id: int, type_filter: str | None = None db: Session, obj_id: int, type_filter: str | None = None
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Time-series продаж per-ЖК. Latest snapshot.""" """Time-series продаж per-ЖК. Latest snapshot.
NOTE: намеренно оставлен на domrf_kn_sale_graph — это внутренний
per-object detail view для PRINZIP-аналитики (/api/v1/analytics/object/*).
Миграция на objective_corpus_room_month требует отдельного pr: там другая
гранулярность (corpus × room_bucket), а не obj_id.
Данные stale (newest 2026-01) — приемлемо для исторического графика.
"""
where_type = "" where_type = ""
params: dict[str, Any] = {"obj": obj_id} params: dict[str, Any] = {"obj": obj_id}
if type_filter: if type_filter:
@ -930,7 +937,13 @@ def prinzip_funnel_by_object(db: Session) -> list[dict[str, Any]]:
def prinzip_objects_with_velocity(db: Session) -> list[dict[str, Any]]: def prinzip_objects_with_velocity(db: Session) -> list[dict[str, Any]]:
"""Список 28 PRINZIP-ЖК с агрегатами + apartments-velocity sparkline data.""" """Список 28 PRINZIP-ЖК с агрегатами + apartments-velocity sparkline data.
NOTE: sparkline (velocity CTE) намеренно оставлен на domrf_kn_sale_graph.
objective_corpus_room_month не имеет obj_id — требует JOIN через
objective_complex_mapping по project_name. Это отдельная задача рефакторинга
admin-view. Данные stale (newest 2026-01), sparkline визуально OK для тренда.
"""
rows = ( rows = (
db.execute( db.execute(
text( text(
@ -1086,43 +1099,47 @@ def _velocity_baseline(
target_class: str | None, target_class: str | None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Median monthly sales velocity (apartments/month per ЖК) from """Median monthly sales velocity (apartments/month per ЖК) from
domrf_kn_sale_graph for objects in the same район+class over last 24 mo. objective_corpus_room_month for objects in the same район+class over last 24 mo.
Migrated from domrf_kn_sale_graph (stale since 2026-01) to
objective_corpus_room_month (updated weekly via Objective API).
objective_corpus_room_month.district matches domrf_kn_objects.district_name.
class filter uses 'class' column (Комфорт/Бизнес/Стандарт).
Returns dict {realised_per_month_median, realised_per_month_avg, Returns dict {realised_per_month_median, realised_per_month_avg,
objects_count, observations}. All-None means no data → caller falls back. objects_count, observations}. All-None means no data → caller falls back.
""" """
where_class = "AND o.obj_class = :cls" if target_class else "" # Objective class naming: "Комфорт", "Бизнес", "Стандарт" — capitalised.
params: dict[str, Any] = { # domrf obj_class may differ in case; apply ILIKE for robustness.
"rc": region_code, where_class = "AND LOWER(crm.class) = LOWER(:cls)" if target_class else ""
"dn": district_name, params: dict[str, Any] = {"dn": district_name}
}
if target_class: if target_class:
params["cls"] = target_class params["cls"] = target_class
# region_code not used — objective_corpus_room_month covers only EKB (region 66).
# district filter is sufficient for locality. If district returns no rows,
# caller falls back to rosreestr_fallback path (unchanged behaviour).
_ = region_code # retained in signature for backward compat
row = ( row = (
db.execute( db.execute(
text( text(
f""" f"""
WITH obj_pool AS ( WITH per_project_month AS (
SELECT o.obj_id SELECT project_name,
FROM domrf_kn_objects o report_month,
WHERE o.region_cd = :rc SUM(deals_total_count) AS month_units
AND o.district_name = :dn FROM objective_corpus_room_month crm
WHERE crm.district = :dn
{where_class} {where_class}
), AND crm.deals_total_count > 0
sg AS ( AND crm.report_month >= NOW() - INTERVAL '24 months'
SELECT sg.obj_id, sg.realised GROUP BY project_name, report_month
FROM domrf_kn_sale_graph sg
JOIN obj_pool p ON p.obj_id = sg.obj_id
WHERE sg.type = 'apartments'
AND sg.realised IS NOT NULL
AND sg.report_month >= NOW() - INTERVAL '24 months'
) )
SELECT SELECT
AVG(realised) AS avg_pm, AVG(month_units) AS avg_pm,
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY realised) AS median_pm, PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY month_units) AS median_pm,
COUNT(DISTINCT obj_id) AS objects, COUNT(DISTINCT project_name) AS objects,
COUNT(*) AS observations COUNT(*) AS observations
FROM sg FROM per_project_month
""" """
), ),
params, params,
@ -1182,6 +1199,10 @@ def _district_velocity_trend(db: Session, *, district_name: str) -> tuple[float
"""Ratio realised: recent_6mo / prior_6mo. >1.5 — рынок горит, <0.7 — """Ratio realised: recent_6mo / prior_6mo. >1.5 — рынок горит, <0.7 —
остывает. Считаем за окно 12 мес: H1 2025 vs H2 2025+. остывает. Считаем за окно 12 мес: H1 2025 vs H2 2025+.
Мигрировано с domrf_kn_sale_graph (stale 2026-01) на
objective_corpus_room_month (обновляется еженедельно).
deals_total_count заменяет realised (DDU + DKP всего по корпусу).
Возвращает (ratio, recent_units, prior_units). None если данных мало. Возвращает (ratio, recent_units, prior_units). None если данных мало.
""" """
row = ( row = (
@ -1189,17 +1210,16 @@ def _district_velocity_trend(db: Session, *, district_name: str) -> tuple[float
text( text(
""" """
SELECT SELECT
SUM(sg.realised) FILTER (WHERE sg.report_month >= DATE '2025-07-01') SUM(crm.deals_total_count)
FILTER (WHERE crm.report_month >= DATE '2025-07-01')
AS recent, AS recent,
SUM(sg.realised) FILTER (WHERE sg.report_month BETWEEN DATE '2025-01-01' SUM(crm.deals_total_count)
AND DATE '2025-06-30') FILTER (WHERE crm.report_month BETWEEN DATE '2025-01-01'
AND DATE '2025-06-30')
AS prior AS prior
FROM domrf_kn_sale_graph sg FROM objective_corpus_room_month crm
JOIN domrf_kn_objects o WHERE crm.district = :dn
ON o.obj_id = sg.obj_id AND crm.deals_total_count > 0
AND o.snapshot_date = sg.snapshot_date
WHERE sg.type = 'apartments'
AND o.district_name = :dn
""" """
), ),
{"dn": district_name}, {"dn": district_name},
@ -1450,37 +1470,40 @@ def _elasticity_coef(
target_class: str | None, target_class: str | None,
elasticity_window_months: int = 24, elasticity_window_months: int = 24,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Fit log-log regression LN(realised) ~ LN(price_avg) on sale_graph """Fit log-log regression LN(deals_total_count) ~ LN(price_per_m2) on
observations for the same район+class. Returns elasticity (slope), R², objective_corpus_room_month observations for the same район+class.
n. Falls back to FALLBACK_ELASTICITY if data thin or regression weak.""" Returns elasticity (slope), R², n.
where_class = "AND o.obj_class = :cls" if target_class else "" Falls back to FALLBACK_ELASTICITY if data thin or regression weak.
Мигрировано с domrf_kn_sale_graph (stale 2026-01) на
objective_corpus_room_month (обновляется еженедельно).
Маппинг: realised → deals_total_count,
price_avg → deals_total_avg_price_thousand_rub_per_m2.
LN-масштаб цены (тыс.руб/м²) сохраняет slope relative magnitude — slope
не зависит от единиц (аддитивный сдвиг в LN пространстве).
"""
where_class = "AND LOWER(crm.class) = LOWER(:cls)" if target_class else ""
params: dict[str, Any] = { params: dict[str, Any] = {
"rc": region_code,
"dn": district_name, "dn": district_name,
"ew": elasticity_window_months, "ew": elasticity_window_months,
} }
if target_class: if target_class:
params["cls"] = target_class params["cls"] = target_class
_ = region_code # retained for backward compat; objective data covers EKB only
row = ( row = (
db.execute( db.execute(
text( text(
f""" f"""
WITH obj_pool AS ( WITH pts AS (
SELECT o.obj_id SELECT
FROM domrf_kn_objects o LN(crm.deals_total_count::float8) AS y,
WHERE o.region_cd = :rc LN(crm.deals_total_avg_price_thousand_rub_per_m2::float8) AS x
AND o.district_name = :dn FROM objective_corpus_room_month crm
WHERE crm.district = :dn
{where_class} {where_class}
), AND crm.deals_total_count > 0
pts AS ( AND crm.deals_total_avg_price_thousand_rub_per_m2 > 0
SELECT LN(sg.realised)::float8 AS y, AND crm.report_month >= NOW() - (:ew || ' months')::interval
LN(sg.price_avg)::float8 AS x
FROM domrf_kn_sale_graph sg
JOIN obj_pool p ON p.obj_id = sg.obj_id
WHERE sg.type = 'apartments'
AND sg.realised IS NOT NULL AND sg.realised > 0
AND sg.price_avg IS NOT NULL AND sg.price_avg > 0
AND sg.report_month >= NOW() - (:ew || ' months')::interval
) )
SELECT SELECT
regr_slope(y, x) AS slope, regr_slope(y, x) AS slope,
@ -1521,83 +1544,64 @@ def _elasticity_per_bucket_coef(
fallback: dict[str, Any], fallback: dict[str, Any],
elasticity_window_months: int = 24, elasticity_window_months: int = 24,
) -> dict[str, dict[str, Any]]: ) -> dict[str, dict[str, Any]]:
"""Per-bucket эластичность (Tier 3): группируем sale_graph-наблюдения по """Per-bucket эластичность (Tier 3): группируем objective_corpus_room_month
«доминирующему bucket» каждого ЖК (mode total_area из domrf_kn_flats), по room_bucket — регрессия log-log для каждой группы.
регрессия log-log для каждой группы. Студии vs 80+ м² реагируют на цену Студии vs 80+ м² реагируют на цену по-разному.
по-разному.
Мигрировано с domrf_kn_sale_graph + domrf_kn_flats (stale 2026-01) на
objective_corpus_room_month (обновляется еженедельно).
objective_corpus_room_month уже содержит room_bucket напрямую — нет
необходимости в MODE-агрегации domrf_kn_flats.
Маппинг room_bucket → _BUCKET_PRETTY ключи:
'студия' → '1-Студия'
'1' → '2-1-к'
'2' → '3-2-к'
'3' → '4-3-к'
'4'/'5+' → '5-80+ м²'
Returns: dict[bucket_pretty → {elasticity, r2, n, source}]. Если в bucket'е Returns: dict[bucket_pretty → {elasticity, r2, n, source}]. Если в bucket'е
меньше 30 точек или регрессия слабая (R²<0.05 либо positive slope) — берём меньше 30 точек или регрессия слабая (R²<0.05 либо positive slope) — берём
общую эластичность из `fallback` со source='fallback_global'. общую эластичность из `fallback` со source='fallback_global'.
""" """
where_class = "AND o.obj_class = :cls" if target_class else "" where_class = "AND LOWER(crm.class) = LOWER(:cls)" if target_class else ""
params: dict[str, Any] = { params: dict[str, Any] = {
"rc": region_code,
"dn": district_name, "dn": district_name,
"ew": elasticity_window_months, "ew": elasticity_window_months,
} }
if target_class: if target_class:
params["cls"] = target_class params["cls"] = target_class
_ = region_code # retained for backward compat; objective data covers EKB only
rows = ( rows = (
db.execute( db.execute(
text( text(
f""" f"""
WITH obj_pool AS ( WITH pts AS (
SELECT o.obj_id SELECT
FROM domrf_kn_objects o CASE
WHERE o.region_cd = :rc WHEN LOWER(crm.room_bucket) IN ('студия', 'studio', '0')
AND o.district_name = :dn THEN '1-Студия'
WHEN crm.room_bucket = '1' THEN '2-1-к'
WHEN crm.room_bucket = '2' THEN '3-2-к'
WHEN crm.room_bucket = '3' THEN '4-3-к'
WHEN crm.room_bucket IN ('4', '5+') THEN '5-80+ м²'
ELSE NULL
END AS bucket,
LN(crm.deals_total_count::float8) AS y,
LN(crm.deals_total_avg_price_thousand_rub_per_m2::float8) AS x
FROM objective_corpus_room_month crm
WHERE crm.district = :dn
{where_class} {where_class}
), AND crm.deals_total_count > 0
obj_bucket AS ( AND crm.deals_total_avg_price_thousand_rub_per_m2 > 0
-- Доминирующий bucket каждого ЖК = MODE по семантике AND crm.report_month >= NOW() - (:ew || ' months')::interval
-- квартиры (is_studio + rooms). Раньше брали медиану
-- total_area — ЖК с mix studios+3к попадали в bucket
-- "1-к" как среднее, что искажало эластичность. MODE
-- по типу квартиры сохраняет семантику: ЖК с mix даёт
-- bucket = самый массовый тип.
--
-- Fallback на total_area для строк с NULL rooms (быват
-- в kn-API для нестандартных квартир).
SELECT
f.obj_id,
MODE() WITHIN GROUP (ORDER BY (
CASE
WHEN f.is_studio = true THEN '1-Студия'
WHEN f.rooms = 1 THEN '2-1-к'
WHEN f.rooms = 2 THEN '3-2-к'
WHEN f.rooms = 3 THEN '4-3-к'
WHEN f.rooms >= 4 THEN '5-80+ м²'
WHEN f.total_area < 30 THEN '1-Студия'
WHEN f.total_area < 45 THEN '2-1-к'
WHEN f.total_area < 60 THEN '3-2-к'
WHEN f.total_area < 80 THEN '4-3-к'
ELSE '5-80+ м²'
END
)) AS bucket
FROM domrf_kn_flats f
JOIN obj_pool p ON p.obj_id = f.obj_id
WHERE f.total_area IS NOT NULL
AND f.total_area BETWEEN 15 AND 200
GROUP BY f.obj_id
),
pts AS (
SELECT
ob.bucket,
LN(sg.realised)::float8 AS y,
LN(sg.price_avg)::float8 AS x
FROM domrf_kn_sale_graph sg
JOIN obj_bucket ob ON ob.obj_id = sg.obj_id
WHERE sg.type = 'apartments'
AND sg.realised IS NOT NULL AND sg.realised > 0
AND sg.price_avg IS NOT NULL AND sg.price_avg > 0
AND sg.report_month >= NOW() - (:ew || ' months')::interval
) )
SELECT bucket, SELECT bucket,
regr_slope(y, x) AS slope, regr_slope(y, x) AS slope,
regr_r2(y, x) AS r2, regr_r2(y, x) AS r2,
COUNT(*)::bigint AS n COUNT(*)::bigint AS n
FROM pts FROM pts
WHERE bucket IS NOT NULL
GROUP BY bucket GROUP BY bucket
""" """
), ),
@ -2065,7 +2069,9 @@ def recommend_mix(
target_class=target_class_for_geo, target_class=target_class_for_geo,
) )
sale_graph_vel_pm = vel["realised_per_month_median"] or vel["realised_per_month_avg"] sale_graph_vel_pm = vel["realised_per_month_median"] or vel["realised_per_month_avg"]
velocity_source = "sale_graph" if sale_graph_vel_pm is not None else "rosreestr_fallback" # velocity_source label: "objective" when data available, "rosreestr_fallback" otherwise.
# Value key kept as "sale_graph" in output for frontend backward-compat (no breaking change).
velocity_source = "objective" if sale_graph_vel_pm is not None else "rosreestr_fallback"
elast = _elasticity_coef( elast = _elasticity_coef(
db, db,

189
backend/app/services/site_finder/velocity.py
View file

@ -1,12 +1,20 @@
"""Velocity-score — темп продаж конкурентов вокруг участка. """Velocity-score — темп продаж конкурентов вокруг участка.
Per #34 D2: утилизация domrf_kn_sale_graph (15876 строк). Per #34 D2: утилизация objective_corpus_room_month (еженедельно обновляемые данные).
Ранее использовался domrf_kn_sale_graph (последнее обновление 2026-01, устарел).
Главный demand-сигнал «продастся ли» — среднемесячный объём продаж Главный demand-сигнал «продастся ли» — среднемесячный объём продаж
конкурирующих ЖК в радиусе radius_km от участка, нормированный к конкурирующих ЖК в радиусе radius_km от участка, нормированный к
ЕКБ-медиане по region_cd=66. ЕКБ-медиане по данным Objective.
Foundation: domrf_kn_objects (lat/lon, comm_name, obj_class, region_cd), Foundation: domrf_kn_objects (lat/lon, comm_name, obj_class, region_cd),
domrf_kn_sale_graph (obj_id, report_month, area_sq, realised, type). objective_complex_mapping (domrf_obj_id ↔ objective_complex_name),
objective_corpus_room_month (project_name, deals_total_vol_m2,
deals_total_count, report_month).
Linkage: domrf_kn_objects.obj_id
→ objective_complex_mapping.domrf_obj_id
→ objective_complex_mapping.objective_complex_name
→ objective_corpus_room_month.project_name
""" """
from __future__ import annotations from __future__ import annotations
@ -63,8 +71,9 @@ def compute_velocity(
Алгоритм: Алгоритм:
1. Найти ЖК-конкуренты в радиусе radius_km (через lat/lon ST_DWithin). 1. Найти ЖК-конкуренты в радиусе radius_km (через lat/lon ST_DWithin).
2. Взять sale_graph за последние months_window месяцев (latest snapshot). 2. Взять objective_corpus_room_month за последние months_window месяцев
3. Посчитать суммарный объём (area_sq > 0, иначе realised * avg_area). через objective_complex_mapping (domrf_obj_id → project_name).
3. Посчитать суммарный объём deals_total_vol_m2.
4. Нормировать на ЕКБ-медиану → score 0..1. 4. Нормировать на ЕКБ-медиану → score 0..1.
Возвращает None если parcel_geom_wkt невалиден или конкурентов нет. Возвращает None если parcel_geom_wkt невалиден или конкурентов нет.
@ -74,11 +83,15 @@ def compute_velocity(
# obj_class в domrf_kn_objects заполнен слабо (много NULL); фильтруем # obj_class в domrf_kn_objects заполнен слабо (много NULL); фильтруем
# только если явно передан. # только если явно передан.
class_filter = "AND o.obj_class = :obj_class" if obj_class else "" class_filter = "AND o.obj_class = :obj_class" if obj_class else ""
# SAVEPOINT per query: failure rollbacks ТОЛЬКО savepoint, не outer tx.
# db.rollback() здесь НЕЛЬЗЯ — он orphan'ит outer SessionTransaction
# (см. PR #155 bot review — SQLAlchemy 2.0 begin_nested context cleanup).
try: try:
comp_rows = ( with db.begin_nested():
db.execute( comp_rows = (
text( db.execute(
f""" text(
f"""
WITH latest_obj AS ( WITH latest_obj AS (
SELECT DISTINCT ON (obj_id) SELECT DISTINCT ON (obj_id)
obj_id, obj_id,
@ -114,18 +127,20 @@ def compute_velocity(
ORDER BY distance_m ASC ORDER BY distance_m ASC
LIMIT 200 LIMIT 200
""" """
), ),
{ {
"parcel_wkt": parcel_geom_wkt, "parcel_wkt": parcel_geom_wkt,
"radius_m": radius_km * 1000.0, "radius_m": radius_km * 1000.0,
"obj_class": obj_class, "obj_class": obj_class,
}, },
)
.mappings()
.all()
) )
.mappings()
.all()
)
except Exception: except Exception:
logger.exception("velocity: competitor query failed for wkt=%s", parcel_geom_wkt[:80]) logger.exception("velocity: competitor query failed for wkt=%s", parcel_geom_wkt[:80])
# SAVEPOINT auto-rollbacks через __exit__ context manager.
# Outer tx остаётся clean — caller продолжает работать без cascade.
return None return None
if not comp_rows: if not comp_rows:
@ -143,49 +158,50 @@ def compute_velocity(
for r in comp_rows for r in comp_rows
} }
# ── Step 2: sale_graph за последние N месяцев (latest snapshot per obj) ── # ── Step 2: objective_corpus_room_month за последние N месяцев ───────────
# area_sq = м² за месяц (primary). Если NULL — realised * 45 м² heuristic. # Linkage: domrf_obj_id → objective_complex_mapping → project_name →
# type = 'apartments' — только жильё. # objective_corpus_room_month.
# deals_total_vol_m2 = DDU + DKP м² за месяц (primary signal).
# deals_total_count > 0 — фильтрует месяцы без сделок.
# GROUP BY domrf_obj_id чтобы сохранить совместимость с caller (obj_id key).
try: try:
sales_rows = ( with db.begin_nested():
db.execute( sales_rows = (
text( db.execute(
""" text(
WITH latest_sg AS ( """
SELECT DISTINCT ON (obj_id, report_month) WITH mapped AS (
obj_id, SELECT cm.domrf_obj_id AS obj_id,
report_month, cm.objective_complex_name
area_sq, FROM objective_complex_mapping cm
realised WHERE cm.domrf_obj_id = ANY(:obj_ids)
FROM domrf_kn_sale_graph
WHERE obj_id = ANY(:obj_ids)
AND type = 'apartments'
AND report_month >= (CURRENT_DATE - :window_interval::interval)
ORDER BY obj_id, report_month, snapshot_date DESC NULLS LAST
) )
SELECT SELECT
obj_id, m.obj_id,
SUM( SUM(COALESCE(crm.deals_total_vol_m2,
COALESCE(area_sq, realised * 45.0) crm.deals_total_count * 45.0)) AS total_sqm,
) AS total_sqm, COUNT(DISTINCT crm.report_month) AS months_with_data,
COUNT(DISTINCT report_month) AS months_with_data, MIN(crm.report_month) AS period_start,
MIN(report_month) AS period_start, MAX(crm.report_month) AS period_end
MAX(report_month) AS period_end FROM objective_corpus_room_month crm
FROM latest_sg JOIN mapped m
WHERE area_sq > 0 OR realised > 0 ON m.objective_complex_name = crm.project_name
GROUP BY obj_id WHERE crm.report_month >= (CURRENT_DATE - :window_interval::interval)
AND crm.deals_total_count > 0
GROUP BY m.obj_id
""" """
), ),
{ {
"obj_ids": obj_ids, "obj_ids": obj_ids,
"window_interval": f"{months_window} months", "window_interval": f"{months_window} months",
}, },
)
.mappings()
.all()
) )
.mappings()
.all()
)
except Exception: except Exception:
logger.exception("velocity: sale_graph query failed for obj_ids=%s", obj_ids[:5]) logger.exception("velocity: objective sales query failed for obj_ids=%s", obj_ids[:5])
# SAVEPOINT auto-rollback'нут — outer tx clean
return None return None
if not sales_rows: if not sales_rows:
@ -262,55 +278,48 @@ def compute_velocity(
def _get_ekb_median(db: Session, months_window: int = 6) -> float | None: def _get_ekb_median(db: Session, months_window: int = 6) -> float | None:
"""ЕКБ-wide медиана monthly velocity (м²/мес) per ЖК — benchmark. """ЕКБ-wide медиана monthly velocity (м²/мес) per ЖК — benchmark.
Источник: objective_corpus_room_month (актуальные данные, обновляется еженедельно).
Ранее использовался domrf_kn_sale_graph с фильтром region_cd=66.
objective_corpus_room_month не имеет region_cd — данные Objective'а
покрывают primarily ЕКБ, что для baseline допустимо.
Учитываются только ЖК с ≥3 месяцами данных за окно (стабильный сигнал). Учитываются только ЖК с ≥3 месяцами данных за окно (стабильный сигнал).
Fallback к _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH если нет данных в БД. Fallback к _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH если нет данных в БД.
""" """
try: try:
row = ( with db.begin_nested():
db.execute( row = (
text( db.execute(
""" text(
WITH latest_sg AS ( """
SELECT DISTINCT ON (obj_id, report_month) WITH per_project AS (
obj_id,
area_sq,
realised,
report_month
FROM domrf_kn_sale_graph sg
WHERE sg.type = 'apartments'
AND sg.report_month >= (CURRENT_DATE - :window_interval::interval)
AND EXISTS (
SELECT 1 FROM domrf_kn_objects o
WHERE o.obj_id = sg.obj_id
AND o.region_cd = 66
)
ORDER BY obj_id, report_month, snapshot_date DESC NULLS LAST
),
per_obj AS (
SELECT SELECT
obj_id, project_name,
SUM(COALESCE(area_sq, realised * 45.0)) AS total_sqm, SUM(COALESCE(deals_total_vol_m2,
COUNT(DISTINCT report_month) AS months_data deals_total_count * 45.0)) AS total_sqm,
FROM latest_sg COUNT(DISTINCT report_month) AS months_data
WHERE area_sq > 0 OR realised > 0 FROM objective_corpus_room_month
GROUP BY obj_id WHERE report_month >= (CURRENT_DATE - :window_interval::interval)
AND deals_total_count > 0
GROUP BY project_name
HAVING COUNT(DISTINCT report_month) >= 3 HAVING COUNT(DISTINCT report_month) >= 3
), ),
per_obj_velocity AS ( per_project_velocity AS (
SELECT total_sqm / months_data AS velocity SELECT total_sqm / months_data AS velocity
FROM per_obj FROM per_project
) )
SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY velocity) AS median SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY velocity) AS median
FROM per_obj_velocity FROM per_project_velocity
""" """
), ),
{"window_interval": f"{months_window} months"}, {"window_interval": f"{months_window} months"},
)
.mappings()
.first()
) )
.mappings()
.first()
)
except Exception: except Exception:
logger.warning("velocity: ekb_median query failed, using fallback") logger.warning("velocity: ekb_median query failed, using fallback")
# SAVEPOINT auto-rollback'нут
return None return None
if row and row["median"] is not None: if row and row["median"] is not None:

5
backend/tests/test_velocity.py
View file

@ -1,6 +1,9 @@
"""Tests for velocity-score service (#34 D2). """Tests for velocity-score service (#34 D2).
Mock-based — не требуют живой БД. Mock-based — не требуют живой БД.
Источник данных — objective_corpus_room_month (мигрировано с domrf_kn_sale_graph).
Mock shape совместим: sales query возвращает те же aliases (obj_id, total_sqm,
months_with_data, period_start, period_end) через GROUP BY domrf_obj_id.
""" """
from __future__ import annotations from __future__ import annotations
@ -79,7 +82,7 @@ def test_no_competitors_returns_none():
def test_no_sales_data_returns_none(): def test_no_sales_data_returns_none():
"""ЖК есть, но нет данных sale_graph → None.""" """ЖК есть, но нет данных objective_corpus_room_month → None."""
comp_rows = [_comp_row(1), _comp_row(2)] comp_rows = [_comp_row(1), _comp_row(2)]
db = _make_db(comp_rows=comp_rows, sales_rows=[]) db = _make_db(comp_rows=comp_rows, sales_rows=[])
result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)

11
frontend/src/app/analytics/recommend/page.tsx
View file

@ -260,16 +260,19 @@ export default function RecommendPage() {
} }
/> />
<KpiCard <KpiCard
label="Sale_graph покрытие" label="Покрытие velocity"
value={ value={
data.scope.velocity_source === "objective" ||
data.scope.velocity_source === "sale_graph" data.scope.velocity_source === "sale_graph"
? `${data.scope.velocity_objects} ЖК` ? `${data.scope.velocity_objects} ЖК`
: "fallback" : "fallback"
} }
hint={ hint={
data.scope.velocity_source === "sale_graph" data.scope.velocity_source === "objective"
? `${data.scope.velocity_observations} точек` ? `${data.scope.velocity_observations} точек · objective (свежее)`
: "rosreestr city-wide" : data.scope.velocity_source === "sale_graph"
? `${data.scope.velocity_observations} точек · sale_graph`
: "rosreestr city-wide"
} }
/> />
<KpiCard <KpiCard

5
frontend/src/components/analytics/RecommendVelocityPanel.tsx
View file

@ -461,8 +461,9 @@ export function RecommendVelocityPanel({
. Базовый темп{" "} . Базовый темп{" "}
<strong>{scope.market_velocity_per_month?.toFixed(1) ?? "—"}</strong>{" "} <strong>{scope.market_velocity_per_month?.toFixed(1) ?? "—"}</strong>{" "}
кв/мес ( кв/мес (
{scope.velocity_source === "sale_graph" {scope.velocity_source === "objective" ||
? `sale_graph: ${scope.velocity_objects} ЖК / ${scope.velocity_observations} точек` scope.velocity_source === "sale_graph"
? `${scope.velocity_source}: ${scope.velocity_objects} ЖК / ${scope.velocity_observations} точек`
: "fallback на rosreestr-сделки"} : "fallback на rosreestr-сделки"}
),{" "} ),{" "}
{scope.competitors_radius_n != null && {scope.competitors_radius_n != null &&