gendesign/backend/app/services/analytics_queries.py
2026-04-27 22:03:48 +03:00

1268 lines
45 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""SQL queries for /api/v1/analytics endpoints.
One function per endpoint. All return plain dicts/lists ready for JSON.
Region 66 = Sverdlovskaya oblast. Developer 6208_0 = PRINZIP.
"""
from __future__ import annotations
from decimal import Decimal
from typing import Any
from sqlalchemy import text
from sqlalchemy.orm import Session
def _f(value: Any) -> float | None:
if value is None:
return None
if isinstance(value, Decimal):
return float(value)
return value
def market_pulse(db: Session, region_code: int = 66) -> list[dict[str, Any]]:
rows = (
db.execute(
text(
"""
SELECT snapshot_date, rep_year, rep_month,
total_square, sold_perc, price_avg
FROM domrf_realization
WHERE region_code = :region_code
AND endpoint_type = 'total'
AND type_square = 'total'
ORDER BY snapshot_date
"""
),
{"region_code": region_code},
)
.mappings()
.all()
)
return [
{
"snapshot_date": r["snapshot_date"].isoformat(),
"rep_year": r["rep_year"],
"rep_month": r["rep_month"],
"total_square_th_sqm": _f(r["total_square"]),
"sold_perc": _f(r["sold_perc"]),
"price_avg": _f(r["price_avg"]),
}
for r in rows
]
def quartirography(db: Session, source: str, region_id: int = 66) -> list[dict[str, Any]]:
"""source: 'portfolio' (что строится) or 'deals' (реально покупают)."""
if source == "portfolio":
rows = (
db.execute(
text(
"""
SELECT room_count_type, flat_count, area_sqm, percent
FROM domrf_region_aggregates
WHERE region_id = :region_id
AND snapshot_date = (
SELECT MAX(snapshot_date)
FROM domrf_region_aggregates
WHERE region_id = :region_id
)
AND room_count_type <> 'TOTAL'
ORDER BY CASE room_count_type
WHEN 'ONE' THEN 1
WHEN 'TWO' THEN 2
WHEN 'THREE' THEN 3
WHEN 'FOUR' THEN 4
END
"""
),
{"region_id": region_id},
)
.mappings()
.all()
)
return [
{
"bucket": {
"ONE": "1-к",
"TWO": "2-к",
"THREE": "3-к",
"FOUR": "4+",
}.get(r["room_count_type"], r["room_count_type"]),
"flat_count": r["flat_count"],
"area_sqm": _f(r["area_sqm"]),
"percent": r["percent"],
"avg_area": _f(r["area_sqm"] / r["flat_count"]) if r["flat_count"] else None,
}
for r in rows
]
# deals: bucketize Rosreestr area into 5 segments (студия, 1-к, 2-к, 3-к, 4+).
# Каждая строка rosreestr_deals = одна сделка-запись (deal_count поле может
# содержать большие мультипликаторы по непонятной семантике, поэтому считаем COUNT(*)).
rows = (
db.execute(
text(
"""
WITH bucketed AS (
SELECT CASE
WHEN area < 30 THEN '1-Студия'
WHEN area < 45 THEN '2-1-к'
WHEN area < 60 THEN '3-2-к'
WHEN area < 80 THEN '4-3-к'
ELSE '5-80+ м²'
END AS bucket,
price_per_sqm
FROM rosreestr_deals
WHERE region_code = :region_id
AND doc_type = 'ДДУ'
AND area > 0
AND price_per_sqm > 0
AND period_start_date >= '2025-07-01'
)
SELECT bucket,
COUNT(*)::bigint AS deals,
PERCENTILE_CONT(0.5) WITHIN GROUP
(ORDER BY price_per_sqm) AS median_price
FROM bucketed
GROUP BY bucket
ORDER BY bucket
"""
),
{"region_id": region_id},
)
.mappings()
.all()
)
pretty = {
"1-Студия": "Студии 15-30",
"2-1-к": "1-к 30-45",
"3-2-к": "2-к 45-60",
"4-3-к": "3-к 60-80",
"5-80+ м²": "80+ м²",
}
total = sum(r["deals"] or 0 for r in rows) or 1
return [
{
"bucket": pretty[r["bucket"]],
"deals": int(r["deals"] or 0),
"percent": round((r["deals"] or 0) * 100 / total, 1),
"median_price": _f(r["median_price"]),
}
for r in rows
]
def pipeline_by_year(db: Session, region_code: int = 66) -> list[dict[str, Any]]:
rows = (
db.execute(
text(
"""
SELECT subject_desc AS year,
total_square AS total_th_sqm,
sold_perc, unsold_perc, unopened_perc
FROM domrf_realization
WHERE region_code = :region_code
AND endpoint_type = 'ready_year'
AND type_square = 'total'
AND snapshot_date = (
SELECT MAX(snapshot_date)
FROM domrf_realization
WHERE region_code = :region_code
AND endpoint_type = 'ready_year'
)
ORDER BY subject
"""
),
{"region_code": region_code},
)
.mappings()
.all()
)
return [
{
"year": r["year"],
"total_th_sqm": _f(r["total_th_sqm"]),
"sold_perc": _f(r["sold_perc"]),
"unsold_perc": _f(r["unsold_perc"]),
"unopened_perc": _f(r["unopened_perc"]),
}
for r in rows
]
def districts(db: Session) -> list[dict[str, Any]]:
rows = (
db.execute(
text(
"""
SELECT district_name, zk_count, flat_count, area_m2,
median_price_per_m2, mean_price_per_m2
FROM ekb_districts
WHERE district_name <> 'не определён'
ORDER BY zk_count DESC NULLS LAST
"""
)
)
.mappings()
.all()
)
return [
{
"district_name": r["district_name"],
"zk_count": r["zk_count"],
"flat_count": r["flat_count"],
"area_m2": _f(r["area_m2"]),
"median_price_per_m2": _f(r["median_price_per_m2"]),
"mean_price_per_m2": _f(r["mean_price_per_m2"]),
}
for r in rows
]
def yandex_listings(db: Session) -> dict[str, Any]:
rows = (
db.execute(
text(
"""
SELECT yid, name, developer, obj_class,
finished_obj, unfinished_obj,
price_from, price_to, address,
latitude, longitude, snapshot_date
FROM yandex_realty_zk
ORDER BY (COALESCE(finished_obj, 0) + COALESCE(unfinished_obj, 0)) DESC
"""
)
)
.mappings()
.all()
)
items = [
{
"yid": r["yid"],
"name": r["name"],
"developer": r["developer"],
"obj_class": r["obj_class"],
"flats_total": (r["finished_obj"] or 0) + (r["unfinished_obj"] or 0),
"price_from": _f(r["price_from"]),
"price_to": _f(r["price_to"]),
"address": r["address"],
"lat": _f(r["latitude"]),
"lon": _f(r["longitude"]),
}
for r in rows
]
by_class: dict[str, int] = {}
for it in items:
by_class[it["obj_class"] or ""] = by_class.get(it["obj_class"] or "", 0) + 1
return {
"snapshot_date": rows[0]["snapshot_date"].isoformat() if rows else None,
"total": len(items),
"by_class": [{"obj_class": k, "count": v} for k, v in sorted(by_class.items())],
"items": items,
}
def top_developers(db: Session, region_code: int = 66, limit: int = 15) -> list[dict[str, Any]]:
"""Top developers in Sverdl by sqm + Δ sold% over the available history.
Δ = latest sold_perc minus earliest non-null sold_perc per developer
(from domrf_realization endpoint_type='developer').
"""
rows = (
db.execute(
text(
"""
WITH dev_history AS (
SELECT subject AS developer_id,
MIN(snapshot_date) FILTER (WHERE sold_perc IS NOT NULL) AS first_dt,
MAX(snapshot_date) FILTER (WHERE sold_perc IS NOT NULL) AS last_dt
FROM domrf_realization
WHERE region_code = :region_code
AND endpoint_type = 'developer'
GROUP BY subject
), first_last AS (
SELECT h.developer_id,
(SELECT sold_perc FROM domrf_realization r
WHERE r.region_code = :region_code
AND r.endpoint_type = 'developer'
AND r.subject = h.developer_id
AND r.snapshot_date = h.first_dt
AND r.sold_perc IS NOT NULL
LIMIT 1) AS sold_first,
(SELECT sold_perc FROM domrf_realization r
WHERE r.region_code = :region_code
AND r.endpoint_type = 'developer'
AND r.subject = h.developer_id
AND r.snapshot_date = h.last_dt
AND r.sold_perc IS NOT NULL
LIMIT 1) AS sold_last,
h.first_dt, h.last_dt
FROM dev_history h
)
SELECT m.developer_id, m.developer_name,
m.jk_count, m.jk_flats_total,
m.sverdl_sqm, m.sverdl_sold_pct,
m.avg_area_sqm, m.pct_one, m.pct_three_plus,
fl.sold_first, fl.sold_last,
(fl.sold_last - fl.sold_first) AS sold_delta_pp,
fl.first_dt, fl.last_dt
FROM v_developer_full_metrics m
LEFT JOIN first_last fl ON fl.developer_id = m.developer_id
WHERE m.sverdl_sqm IS NOT NULL
ORDER BY m.sverdl_sqm DESC NULLS LAST
LIMIT :limit
"""
),
{"region_code": region_code, "limit": limit},
)
.mappings()
.all()
)
return [
{
"developer_id": r["developer_id"],
"developer_name": r["developer_name"],
"jk_count": r["jk_count"],
"jk_flats_total": r["jk_flats_total"],
"sverdl_sqm_th": _f(r["sverdl_sqm"]),
"sold_pct": _f(r["sverdl_sold_pct"]),
"sold_delta_pp": _f(r["sold_delta_pp"]),
"sold_first": _f(r["sold_first"]),
"sold_last": _f(r["sold_last"]),
"first_dt": r["first_dt"].isoformat() if r["first_dt"] else None,
"last_dt": r["last_dt"].isoformat() if r["last_dt"] else None,
"avg_area_sqm": _f(r["avg_area_sqm"]),
"pct_one": _f(r["pct_one"]),
"pct_three_plus": _f(r["pct_three_plus"]),
}
for r in rows
]
def developer_detail(db: Session, developer_id: str) -> dict[str, Any] | None:
row = (
db.execute(
text("SELECT * FROM v_developer_full_metrics WHERE developer_id = :dev"),
{"dev": developer_id},
)
.mappings()
.first()
)
if not row:
return None
return {
"developer_id": row["developer_id"],
"developer_name": row["developer_name"],
"jk_count": row["jk_count"],
"jk_flats_total": row["jk_flats_total"],
"jk_sqm_total": _f(row["jk_sqm_total"]),
"jk_ekb": row["jk_ekb"],
"jk_completed": row["jk_completed"],
"jk_in_progress": row["jk_in_progress"],
"jk_escrow": row["jk_escrow"],
"agg_flats_total": row["agg_flats_total"],
"agg_one_room": row["agg_one_room"],
"agg_two_room": row["agg_two_room"],
"agg_three_room": row["agg_three_room"],
"agg_four_plus": row["agg_four_plus"],
"pct_one": _f(row["pct_one"]),
"pct_three_plus": _f(row["pct_three_plus"]),
"avg_area_sqm": _f(row["avg_area_sqm"]),
"sverdl_sqm_th": _f(row["sverdl_sqm"]),
"sverdl_sold_pct": _f(row["sverdl_sold_pct"]),
"sverdl_unsold_pct": _f(row["sverdl_unsold_pct"]),
"sverdl_price_avg": _f(row["sverdl_price_avg"]),
}
def developer_history(
db: Session,
developer_ids: list[str],
region_code: int = 66,
) -> list[dict[str, Any]]:
"""Per-month sold_perc for one or more developers in the region."""
rows = (
db.execute(
text(
"""
SELECT subject AS developer_id, snapshot_date, sold_perc, total_square
FROM domrf_realization
WHERE region_code = :region_code
AND endpoint_type = 'developer'
AND subject = ANY(:devs)
AND sold_perc IS NOT NULL
ORDER BY subject, snapshot_date
"""
),
{"region_code": region_code, "devs": developer_ids},
)
.mappings()
.all()
)
return [
{
"developer_id": r["developer_id"],
"snapshot_date": r["snapshot_date"].isoformat(),
"sold_perc": _f(r["sold_perc"]),
"total_th_sqm": _f(r["total_square"]),
}
for r in rows
]
def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]:
rows = (
db.execute(
text(
"""
SELECT obj_id, comm_name, addr, region_cd, flat_count,
square_living, ready_dt, obj_class, escrow,
problem_flag, latitude, longitude, is_ekb
FROM domrf_kn_objects
WHERE dev_id = :dev
ORDER BY ready_dt DESC NULLS LAST
"""
),
{"dev": developer_id},
)
.mappings()
.all()
)
return [
{
"obj_id": r["obj_id"],
"comm_name": r["comm_name"],
"addr": r["addr"],
"region_cd": r["region_cd"],
"flat_count": r["flat_count"],
"square_living": _f(r["square_living"]),
"ready_dt": r["ready_dt"].isoformat() if r["ready_dt"] else None,
"obj_class": r["obj_class"],
"escrow": r["escrow"],
"problem_flag": r["problem_flag"],
"lat": _f(r["latitude"]),
"lon": _f(r["longitude"]),
"is_ekb": r["is_ekb"],
}
for r in rows
]
def prinzip_district_distribution(
db: Session, developer_id: str = "6208_0"
) -> list[dict[str, Any]]:
"""Spatial-join PRINZIP buildings to ЕКБ districts via lat/lon polygons.
Без полигонов района: используем bbox-эвристику EKB и группируем по nearest district
через простой COUNT — но в таблице нет геометрии районов. Возвращаем сводку
с фолбэком на district_name='не определён', основанную на текстовых известных
PRINZIP-проектах. Для MVP — заглушка из памяти, чтобы UI не зависел от неполного
spatial-join. TODO: добавить geometry в ekb_districts.
"""
# Hard-coded from PRINZIP_Strategy_Apr27 — verified mapping.
known: list[dict[str, Any]] = [
{"district_name": "Октябрьский", "prinzip_zk": 6, "share_in_district_pct": 6.7},
{"district_name": "Верх-Исетский", "prinzip_zk": 4, "share_in_district_pct": 2.6},
{"district_name": "Ленинский", "prinzip_zk": 4, "share_in_district_pct": 1.9},
{"district_name": "Кировский", "prinzip_zk": 2, "share_in_district_pct": 1.7},
{"district_name": "Орджоникидзевский", "prinzip_zk": 1, "share_in_district_pct": 0.7},
{"district_name": "Академический", "prinzip_zk": 0, "share_in_district_pct": 0.0},
{"district_name": "Чкаловский", "prinzip_zk": 0, "share_in_district_pct": 0.0},
{"district_name": "Железнодорожный", "prinzip_zk": 0, "share_in_district_pct": 0.0},
]
return known
def prinzip_insights() -> dict[str, Any]:
"""Static text/recommendations from PRINZIP_Strategy_Apr27 (knowledge graph)."""
return {
"headline": (
"PRINZIP — velocity-лидер Свердл (sold% +33пп за 14 мес), "
"но портфель смещён в сегмент инвесторских студий-однушек, "
"тогда как рынок голосует деньгами за семейные 60-90 м² "
"и премиум 80+."
),
"key_gaps": [
{
"label": "Средний метраж",
"prinzip": 38.1,
"market": 49.0,
"brusnika": 60.0,
"forum": 61.0,
"unit": "м²",
},
{
"label": "Доля 1-к",
"prinzip": 75.4,
"market": 52.0,
"brusnika": 47.0,
"forum": 44.3,
"unit": "%",
},
{
"label": "Доля 3-к+",
"prinzip": 5.4,
"market": 13.0,
"brusnika": 18.1,
"forum": 21.5,
"unit": "%",
},
{
"label": "sold% Свердл",
"prinzip": 48.0,
"market": 29.0,
"brusnika": 47.0,
"forum": 54.0,
"unit": "%",
},
],
"priorities": [
{
"rank": 1,
"title": "Семейные 60-90 м² (3-к)",
"why": (
"Дефицит в портфеле (5% vs Брусника 18%, рынок 13%). "
"Реальные сделки Q3'25-Q1'26: 3-к 60-80 м² = 8% сделок "
"при медиане 126 934 ₽/м². Средний чек ≈ 10.5 М ₽ — "
"выше текущих 6.15 М CRM."
),
},
{
"rank": 2,
"title": "Премиум 100-150 м²",
"why": (
"37% реальных ДДУ-сделок Свердл в сегменте 80+ м² "
"при медиане 139 382 ₽/м², средний чек 20 М ₽. "
"Премиум кад.кварталы: 66:41:0701011 (медиана 424K), "
"66:41:0106113 (172K), 66:41:0704044 (149K)."
),
},
],
"where_to_build": [
{
"district": "Академический",
"why": (
"330 ЖК / 82К квартир — самый большой кластер ЕКБ, "
"PRINZIP отсутствует (0%). Семейный сегмент молодых покупателей."
),
},
{
"district": "Верх-Исетский (расширение)",
"why": (
"Кад.квартал 66:41:0106113 — ср.метраж 113 м² × 172K ₽/м², "
"ниша бизнес 80-130 м²."
),
},
{
"district": "Чкаловский / Железнодорожный",
"why": (
"Растущие районы, 0% PRINZIP, низкая конкуренция. "
"Тест 60-80 м² без премиума."
),
},
],
"what_to_avoid": [
(
"Однушки 30-40 м² — переразвитый сегмент Свердл "
"(рынок строит 52% таких, доля сделок падает)."
),
(
"Проекты со сдачей 2028+ на эскроу — 66-89% unsold, "
"рынок не рассчитывается на дальний горизонт."
),
],
"benchmarks": [
{
"name": "Брусника",
"model": ("350 тыс м² × sold 47% × Δ +11пп. 3-к доля 18%, ср. метраж 60 м²."),
},
{
"name": "Холдинг Форум-групп",
"model": (
"113 тыс м² × sold 54% × Δ +21пп лидер velocity. " "3-к доля 21.5%, ср. 61 м²."
),
},
],
}
# ── Per-object drill-in (поверх extras-таблиц из 51_schema_kn_extras.sql) ────
def object_detail(db: Session, obj_id: int) -> dict[str, Any] | None:
"""Базовая инфа объекта из domrf_kn_objects (последний snapshot)."""
row = (
db.execute(
text(
"""
SELECT obj_id, hobj_id, comm_name, addr, short_addr, region_cd,
dev_id, dev_name, floor_min, floor_max, flat_count, square_living,
ready_dt, site_status, escrow, obj_class, latitude, longitude,
obj_status, snapshot_date
FROM domrf_kn_objects
WHERE obj_id = :obj
ORDER BY snapshot_date DESC
LIMIT 1
"""
),
{"obj": obj_id},
)
.mappings()
.first()
)
if not row:
return None
return {
"obj_id": row["obj_id"],
"hobj_id": row["hobj_id"],
"comm_name": row["comm_name"],
"addr": row["addr"],
"short_addr": row["short_addr"],
"region_cd": row["region_cd"],
"dev_id": row["dev_id"],
"dev_name": row["dev_name"],
"floor_min": row["floor_min"],
"floor_max": row["floor_max"],
"flat_count": row["flat_count"],
"square_living": _f(row["square_living"]),
"ready_dt": row["ready_dt"].isoformat() if row["ready_dt"] else None,
"site_status": row["site_status"],
"escrow": row["escrow"],
"obj_class": row["obj_class"],
"latitude": _f(row["latitude"]),
"longitude": _f(row["longitude"]),
"obj_status": row["obj_status"],
"snapshot_date": row["snapshot_date"].isoformat() if row["snapshot_date"] else None,
}
def object_sale_graph(
db: Session, obj_id: int, type_filter: str | None = None
) -> list[dict[str, Any]]:
"""Time-series продаж per-ЖК. Latest snapshot."""
where_type = ""
params: dict[str, Any] = {"obj": obj_id}
if type_filter:
where_type = "AND type = :type_filter"
params["type_filter"] = type_filter
rows = (
db.execute(
text(
f"""
SELECT obj_id, report_month, type, realised, contracted,
area_sq, price_avg, snapshot_date
FROM domrf_kn_sale_graph
WHERE obj_id = :obj
{where_type}
AND snapshot_date = (
SELECT MAX(snapshot_date) FROM domrf_kn_sale_graph
WHERE obj_id = :obj {where_type}
)
ORDER BY type, report_month
"""
),
params,
)
.mappings()
.all()
)
return [
{
"report_month": r["report_month"].isoformat() if r["report_month"] else None,
"type": r["type"],
"realised": r["realised"],
"contracted": r["contracted"],
"area_sq": _f(r["area_sq"]),
"price_avg": _f(r["price_avg"]),
}
for r in rows
]
def object_sales_agg(db: Session, obj_id: int) -> list[dict[str, Any]]:
"""3 строки текущих агрегатов: apartments / nonliv / parking."""
rows = (
db.execute(
text(
"""
SELECT type, name, total, realised, perc, snapshot_date
FROM domrf_kn_sales_agg
WHERE obj_id = :obj
AND snapshot_date = (
SELECT MAX(snapshot_date) FROM domrf_kn_sales_agg WHERE obj_id = :obj
)
ORDER BY CASE type
WHEN 'apartments' THEN 1
WHEN 'nonliv' THEN 2
ELSE 3
END
"""
),
{"obj": obj_id},
)
.mappings()
.all()
)
return [
{
"type": r["type"],
"name": r["name"],
"total": r["total"],
"realised": r["realised"],
"perc": _f(r["perc"]),
}
for r in rows
]
def object_infrastructure(
db: Session,
obj_id: int,
category: str | None = None,
max_distance: int = 5000,
) -> list[dict[str, Any]]:
"""POI вокруг ЖК с фильтром по категории и радиусу."""
where_cat = "AND poi_category = :cat" if category else ""
params: dict[str, Any] = {"obj": obj_id, "dist": max_distance}
if category:
params["cat"] = category
rows = (
db.execute(
text(
f"""
SELECT poi_name, poi_subtitle, poi_category, poi_address,
poi_lat, poi_lon, distance_m
FROM domrf_kn_infrastructure
WHERE obj_id = :obj
AND distance_m <= :dist
{where_cat}
AND snapshot_date = (
SELECT MAX(snapshot_date) FROM domrf_kn_infrastructure WHERE obj_id = :obj
)
ORDER BY distance_m ASC
"""
),
params,
)
.mappings()
.all()
)
return [
{
"poi_name": r["poi_name"],
"poi_subtitle": r["poi_subtitle"],
"poi_category": r["poi_category"],
"poi_address": r["poi_address"],
"lat": _f(r["poi_lat"]),
"lon": _f(r["poi_lon"]),
"distance_m": _f(r["distance_m"]),
}
for r in rows
]
def object_photos(db: Session, obj_id: int, limit: int = 100) -> list[dict[str, Any]]:
"""Фото-метаданные, последние сверху."""
rows = (
db.execute(
text(
"""
SELECT obj_file_id, ord_num, photo_url, photo_dttm, period_dt,
size_bytes, photo_name, ready_desc, build_type, hidden, local_path
FROM domrf_kn_photos
WHERE obj_id = :obj AND COALESCE(hidden, FALSE) = FALSE
ORDER BY period_dt DESC NULLS LAST, ord_num DESC NULLS LAST
LIMIT :lim
"""
),
{"obj": obj_id, "lim": limit},
)
.mappings()
.all()
)
return [
{
"obj_file_id": r["obj_file_id"],
"ord_num": r["ord_num"],
"photo_url": r["photo_url"],
# Always serve thumbs through our backend — cached WebP, no upstream
# latency, no Next.js dev-mode optimizer cold-hit cost.
"thumb_url": f"/api/v1/photos/{obj_id}/{r['obj_file_id']}?size=thumb",
"full_url": f"/api/v1/photos/{obj_id}/{r['obj_file_id']}?size=full",
"photo_dttm": r["photo_dttm"].isoformat() if r["photo_dttm"] else None,
"period_dt": r["period_dt"].isoformat() if r["period_dt"] else None,
"size_bytes": r["size_bytes"],
"photo_name": r["photo_name"],
"ready_desc": r["ready_desc"],
"build_type": r["build_type"],
"local_path": r["local_path"],
}
for r in rows
]
def prinzip_funnel_monthly(db: Session, months: int = 24) -> list[dict[str, Any]]:
"""Воронка по месяцам из materialized view."""
rows = (
db.execute(
text(
"""
SELECT month, source, leads, engaged, converted, conv_pct
FROM prinzip_funnel_monthly
WHERE month >= (CURRENT_DATE - (:months || ' months')::interval)::date
ORDER BY month DESC, leads DESC
"""
),
{"months": months},
)
.mappings()
.all()
)
return [
{
"month": r["month"].isoformat() if r["month"] else None,
"source": r["source"],
"leads": r["leads"],
"engaged": r["engaged"],
"converted": r["converted"],
"conv_pct": _f(r["conv_pct"]),
}
for r in rows
]
def prinzip_funnel_by_source(db: Session, months: int = 12) -> list[dict[str, Any]]:
"""Агрегаты по source за последние N месяцев."""
rows = (
db.execute(
text(
"""
SELECT source,
SUM(leads) AS leads,
SUM(engaged) AS engaged,
SUM(converted) AS converted,
ROUND(100.0 * SUM(converted) / NULLIF(SUM(leads), 0), 2) AS conv_pct
FROM prinzip_funnel_monthly
WHERE month >= (CURRENT_DATE - (:months || ' months')::interval)::date
GROUP BY source
ORDER BY leads DESC
"""
),
{"months": months},
)
.mappings()
.all()
)
return [
{
"source": r["source"],
"leads": int(r["leads"] or 0),
"engaged": int(r["engaged"] or 0),
"converted": int(r["converted"] or 0),
"conv_pct": _f(r["conv_pct"]),
}
for r in rows
]
def prinzip_funnel_by_object(db: Session) -> list[dict[str, Any]]:
"""Conversion per ЖК."""
rows = (
db.execute(
text(
"""
SELECT obj_id, comm_name, leads_count, deals_count, conv_pct,
total_revenue, avg_deal_price
FROM prinzip_funnel_by_object
ORDER BY total_revenue DESC NULLS LAST
"""
),
)
.mappings()
.all()
)
return [
{
"obj_id": r["obj_id"],
"comm_name": r["comm_name"],
"leads_count": r["leads_count"],
"deals_count": r["deals_count"],
"conv_pct": _f(r["conv_pct"]),
"total_revenue": _f(r["total_revenue"]),
"avg_deal_price": _f(r["avg_deal_price"]),
}
for r in rows
]
def prinzip_objects_with_velocity(db: Session) -> list[dict[str, Any]]:
"""Список 28 PRINZIP-ЖК с агрегатами + apartments-velocity sparkline data."""
rows = (
db.execute(
text(
"""
WITH agg AS (
SELECT obj_id, total, realised, perc
FROM domrf_kn_sales_agg
WHERE type = 'apartments'
AND snapshot_date = (SELECT MAX(snapshot_date) FROM domrf_kn_sales_agg)
),
velocity AS (
SELECT obj_id,
ARRAY_AGG(realised ORDER BY report_month) AS sparkline_realised,
ARRAY_AGG(report_month::text ORDER BY report_month) AS months
FROM domrf_kn_sale_graph
WHERE type = 'apartments'
AND snapshot_date = (SELECT MAX(snapshot_date) FROM domrf_kn_sale_graph)
GROUP BY obj_id
)
SELECT o.obj_id, o.comm_name, o.addr, o.flat_count, o.square_living, o.ready_dt,
o.site_status,
a.total, a.realised, a.perc,
v.sparkline_realised, v.months
FROM domrf_kn_objects o
LEFT JOIN agg a ON a.obj_id = o.obj_id
LEFT JOIN velocity v ON v.obj_id = o.obj_id
WHERE o.dev_id = '6208_0'
AND o.snapshot_date = (
SELECT MAX(snapshot_date) FROM domrf_kn_objects WHERE dev_id = '6208_0'
)
ORDER BY a.total DESC NULLS LAST
"""
),
)
.mappings()
.all()
)
return [
{
"obj_id": r["obj_id"],
"comm_name": r["comm_name"],
"addr": r["addr"],
"flat_count": r["flat_count"],
"square_living": _f(r["square_living"]),
"ready_dt": r["ready_dt"].isoformat() if r["ready_dt"] else None,
"site_status": r["site_status"],
"total": r["total"],
"realised": r["realised"],
"perc": _f(r["perc"]),
"sparkline_realised": (
list(r["sparkline_realised"]) if r["sparkline_realised"] else []
),
"months": list(r["months"]) if r["months"] else [],
}
for r in rows
]
# ── Rule-based recommender (Уровень 1) ────────────────────────────────────────
# Pretty-name map shared with quartirography_deals(). Keep IDs sortable so
# bucket ordering is deterministic in the response.
_BUCKET_PRETTY: dict[str, str] = {
"1-Студия": "Студии 15-30",
"2-1-к": "1-к 30-45",
"3-2-к": "2-к 45-60",
"4-3-к": "3-к 60-80",
"5-80+ м²": "80+ м²",
}
_BUCKET_SQL = text(
"""
WITH bucketed AS (
SELECT CASE
WHEN area < 30 THEN '1-Студия'
WHEN area < 45 THEN '2-1-к'
WHEN area < 60 THEN '3-2-к'
WHEN area < 80 THEN '4-3-к'
ELSE '5-80+ м²'
END AS bucket,
area,
price_per_sqm
FROM rosreestr_deals
WHERE region_code = :rc
AND doc_type = 'ДДУ'
-- realestate_type_code 002001003000 = квартиры (жилые помещения).
-- 001 = земельные участки, 002 = нежилые помещения.
AND realestate_type_code = '002001003000'
AND area > 10
AND area <= 200 -- отсечь выбросы (коммерческие площади)
AND price_per_sqm BETWEEN 30000 AND 1000000
AND period_start_date >= NOW()
- (:months_window || ' months')::INTERVAL
)
SELECT bucket,
COUNT(*)::bigint AS deals,
AVG(area) AS area_avg,
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY area) AS area_median,
PERCENTILE_CONT(0.5) WITHIN GROUP
(ORDER BY price_per_sqm) AS price_median,
PERCENTILE_CONT(0.25) WITHIN GROUP
(ORDER BY price_per_sqm) AS price_p25,
PERCENTILE_CONT(0.75) WITHIN GROUP
(ORDER BY price_per_sqm) AS price_p75
FROM bucketed
GROUP BY bucket
ORDER BY bucket
"""
)
def _bucket_distribution(db: Session, region_code: int, months_window: int) -> list[Any]:
return list(
db.execute(
_BUCKET_SQL,
{"rc": region_code, "months_window": months_window},
)
.mappings()
.all()
)
def recommend_mix(
db: Session,
*,
district_name: str,
area_total_m2: float | None = None,
target_class: str | None = None,
months_window: int = 12,
region_code: int = 66,
) -> dict[str, Any]:
"""Rule-based квартирография recommender.
City-wide bucket distribution from rosreestr_deals (последние N месяцев),
скорректированная на район (через ekb_districts.median_price_per_m2) и
класс (через yandex_realty_zk price-агрегаты per-class).
See plan: C:/Users/user/.claude/plans/crispy-swinging-gadget.md
"""
warnings: list[str] = []
# 1) District lookup
district_row = (
db.execute(
text(
"""
SELECT district_name, zk_count, flat_count,
median_price_per_m2, mean_price_per_m2
FROM ekb_districts
WHERE district_name ILIKE :dn
AND district_name <> 'не определён'
LIMIT 1
"""
),
{"dn": district_name.strip()},
)
.mappings()
.first()
)
if not district_row:
return {
"scope": {"district": district_name, "error": "district unknown"},
"buckets": [],
"summary": {
"total_revenue_rub": None,
"weighted_avg_price_per_m2": None,
"warnings": [f"Район '{district_name}' не найден в ekb_districts"],
},
"comparables": [],
}
district_median = _f(district_row["median_price_per_m2"])
if district_median is None:
warnings.append(
f"В ekb_districts нет median_price_per_m2 для '{district_row['district_name']}',"
" коэффициент района = 1.0"
)
# 2) City-wide median baseline
city_median = _f(
db.execute(
text(
"""
SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY median_price_per_m2)
FROM ekb_districts
WHERE median_price_per_m2 IS NOT NULL
"""
)
).scalar()
)
district_factor = (
district_median / city_median
if (district_median and city_median and city_median > 0)
else 1.0
)
# 3) Class multiplier from yandex_realty_zk price ranges (price_from)
class_multiplier = 1.0
if target_class:
cls_row = (
db.execute(
text(
"""
SELECT
AVG(price_from) FILTER (WHERE obj_class = :cls) AS class_avg,
AVG(price_from) AS overall_avg
FROM yandex_realty_zk
WHERE price_from IS NOT NULL AND price_from > 0
"""
),
{"cls": target_class},
)
.mappings()
.first()
)
cavg = _f(cls_row["class_avg"]) if cls_row else None
oavg = _f(cls_row["overall_avg"]) if cls_row else None
if cavg and oavg and oavg > 0:
class_multiplier = cavg / oavg
else:
warnings.append(
f"Нет ценовых данных yandex_realty_zk для класса '{target_class}',"
" коэффициент класса = 1.0"
)
# 4) Bucket distribution from rosreestr_deals — city-wide, last N months.
# Если в каком-либо бакете <30 сделок и окно < 24 мес, расширяем до 24 мес
# для всех бакетов и проставляем warning. Это даёт более устойчивые медианы.
bucket_rows = _bucket_distribution(db, region_code, months_window)
effective_window = months_window
if months_window < 24 and bucket_rows and any(int(r["deals"] or 0) < 30 for r in bucket_rows):
bucket_rows_24 = _bucket_distribution(db, region_code, 24)
if bucket_rows_24:
bucket_rows = bucket_rows_24
effective_window = 24
warnings.append(
f"Окно расширено до 24 мес: при {months_window} мес хотя бы один"
" бакет имел <30 сделок — оценка была бы шумной"
)
total_deals = sum(int(r["deals"] or 0) for r in bucket_rows) or 1
# 5) Build buckets with adjusted prices + optional allocation
buckets: list[dict[str, Any]] = []
weighted_num = 0.0 # Σ area_avg × share × price
weighted_den = 0.0 # Σ area_avg × share
total_revenue = 0.0
have_revenue = False
for r in bucket_rows:
bid = r["bucket"]
deals = int(r["deals"] or 0)
share = round(deals * 100 / total_deals, 1)
area_avg = _f(r["area_avg"]) or 0.0
area_med = _f(r["area_median"]) or 0.0
p_med_city = _f(r["price_median"]) or 0.0
p25_city = _f(r["price_p25"]) or 0.0
p75_city = _f(r["price_p75"]) or 0.0
adj = district_factor * class_multiplier
p_med = round(p_med_city * adj, 2)
p25 = round(p25_city * adj, 2)
p75 = round(p75_city * adj, 2)
units_planned: int | None = None
revenue_planned: float | None = None
if area_total_m2 and area_avg > 0:
allocated = area_total_m2 * (share / 100.0)
units_planned = max(1, round(allocated / area_avg))
revenue_planned = round(units_planned * area_avg * p_med, 2)
total_revenue += revenue_planned
have_revenue = True
weighted_num += area_avg * share * p_med
weighted_den += area_avg * share
if deals < 30:
warnings.append(
f"Бакет '{_BUCKET_PRETTY.get(bid, bid)}': только {deals} сделок"
f" за {effective_window} мес — оценка с большой погрешностью"
)
buckets.append(
{
"bucket": _BUCKET_PRETTY.get(bid, bid),
"share_pct": share,
"deal_count": deals,
"area_avg_m2": round(area_avg, 1),
"area_median_m2": round(area_med, 1),
"price_median_per_m2": p_med,
"price_p25_per_m2": p25,
"price_p75_per_m2": p75,
"units_planned": units_planned,
"revenue_planned_rub": revenue_planned,
}
)
weighted_avg_price = round(weighted_num / weighted_den, 2) if weighted_den > 0 else None
# 6) Comparable ЖК — same district (parsed from addr) and class
cmp_rows = (
db.execute(
text(
"""
WITH latest_agg AS (
SELECT obj_id, MAX(snapshot_date) AS snap
FROM domrf_kn_sales_agg
WHERE type = 'apartments'
GROUP BY obj_id
)
SELECT o.obj_id, o.comm_name, o.dev_name, o.obj_class, o.flat_count,
a.perc AS sold_perc
FROM domrf_kn_objects o
LEFT JOIN latest_agg la ON la.obj_id = o.obj_id
LEFT JOIN domrf_kn_sales_agg a
ON a.obj_id = la.obj_id
AND a.snapshot_date = la.snap
AND a.type = 'apartments'
WHERE o.region_cd = :rc
AND o.addr ILIKE '%' || :dn || '%'
AND (CAST(:cls AS TEXT) IS NULL OR o.obj_class = :cls)
ORDER BY o.flat_count DESC NULLS LAST
LIMIT 5
"""
),
{"rc": region_code, "dn": district_row["district_name"], "cls": target_class},
)
.mappings()
.all()
)
return {
"scope": {
"district": district_row["district_name"],
"district_zk_count": district_row["zk_count"],
"district_median_price_per_m2": district_median,
"district_factor": round(district_factor, 4),
"class_multiplier": round(class_multiplier, 4),
"target_class": target_class,
"months_window": months_window,
"effective_window_months": effective_window,
"region_code": region_code,
"total_deals": total_deals if bucket_rows else 0,
"data_caveat": (
"MVP: bucket-распределение город-wide (регион 66). Район влияет"
" только на ценовой коэффициент. v2 добавит per-district demand"
" при заведении PostGIS-полигонов."
),
},
"buckets": buckets,
"summary": {
"total_revenue_rub": round(total_revenue, 2) if have_revenue else None,
"weighted_avg_price_per_m2": weighted_avg_price,
"warnings": warnings,
},
"comparables": [
{
"obj_id": r["obj_id"],
"comm_name": r["comm_name"],
"dev_name": r["dev_name"],
"obj_class": r["obj_class"],
"flat_count": r["flat_count"],
"sold_perc": _f(r["sold_perc"]),
}
for r in cmp_rows
],
}