Light1YT
e67cb721bf
objective_lots — current-state UPSERT (UNIQUE objective_lot_id, 5 snapshot_date), но Объектив присваивает ОДНОМУ физлоту (project,corpus,section,floor,lot_number) несколько lot_id за пере-листинги → таблица раздута ~2.91× (прод: 1.75M квартир- строк vs 603k физлотов). История — в отдельной objective_lots_history (не трогаем). STEP 1: миграция 175 — VIEW v_objective_lots_latest (DISTINCT ON physflat-ключ, последний снапшот). ol.* стабилен (51==51 колонок). DRY-RUN на проде: 603 049 квартир vs 1 753 283 raw. STEP 2: репойнт current-state консьюмеров objective_lots → v_objective_lots_latest: - supply_layers._L1_OPEN_SQL (L1 открытое предложение → дефицит-форсайт; прод: Юго-Западный комфорт 58 606 → 12 620) - competitors._SOLD_COUNT_SQL (+ комментарий: COUNT(DISTINCT lot_id) СОХРАНЁН для fan-out-защиты маппинга, не COUNT(*) — view гарантирует physflat-дедуп, DISTINCT гарантирует mapping-fan-out-safety) - parcels.py obj_pricing CTE (карточка конкурента units_sold/available) - special_indices._ARTIFICIAL_DEMAND_SQL - parcels.py district price block + geo-radius median (sample_size/n) - concepts._OBJECTIVE_MEDIAN_SQL (гейт n≥10) - landing KPI3 (% квартир с ценой) - admin_scrape coverage: `lots` оставлен сырым (ETL-fidelity vs SQLite), добавлен `lots_physflat` #1959 inline-дедуп в market_metrics НЕ рефакторим — добавлен комментарий об общем physflat-ключе с view. STEP 3: report_assembler._deal_count теперь = unit_velocity × window_months (оконные продажи), НЕ кумулятивный n_sold. confidence_engine помечает фактор «за 6 мес» и гейтит порогами окна (high≥50) — кумулятив (прод EKB ~380 921) делал гейт бессмысленным и подпись лживой; оконное (~24 876 за 6 мес) честно. Тесты: +guard'ы (L1/sold-count читают view, deal_count оконный). 963 passed.
718 lines
39 KiB
Python
718 lines
39 KiB
Python
"""Market-metrics service — детерминированные рыночные метрики из данных Объектива.
|
||
|
||
#949 PR A (Site Finder v2 / GG-форсайт, EPIC «релевантность конкурентов +
|
||
рыночные метрики»). Это **измерительный слой** (ТЗ §9.2), который потребляют
|
||
forecasting-эпики (#950/#952) и relevance-модель (#949 PR B).
|
||
|
||
Принцип: **детерминированно, без LLM** — чистый set-based SQL + арифметика.
|
||
|
||
Источники (см. `data/sql/68_schema_objective.sql`):
|
||
- `objective_lots` — per-flat текущее состояние (status, is_sold, area_pd,
|
||
rooms_int, district, premise_kind, sales_start_date) +
|
||
`contract_date` («Дата договора») — реальная дата сделки,
|
||
заполнена у 100% проданных лотов. Используем её и для
|
||
кумулятивного стока, и для продаж-в-окне (velocity /
|
||
absorption / sell-through). `objective_lots_history`
|
||
(weekly-снапшоты) НЕ используем: глубина ~17 дней не даёт
|
||
корректного окна продаж (см. #949).
|
||
- elasticity (price_sensitivity) — переиспользуем
|
||
`analytics_queries._elasticity_coef` (objective_corpus_room_month, log-log регрессия).
|
||
|
||
Фильтрация по `district` / `obj_ids` (а НЕ по domrf↔objective маппингу): маппинг
|
||
покрывает ~2.5% объектов, тогда как `district` заполнен у большинства лотов. Это
|
||
обходит mapping-gap — главный риск проекта (sparse coverage).
|
||
|
||
Graceful-on-thin-data (КРИТИЧНО): любая метрика при отсутствии данных = `None`
|
||
(НЕ 0, НЕ crash), `confidence='low'`, результат всё равно возвращается. Каждый
|
||
helper защищён от деления на ноль и пустых выборок.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections.abc import Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from typing import Any, Literal
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.analytics_queries import _elasticity_coef
|
||
from app.services.forecast_request_cache import cached
|
||
from app.services.site_finder.district_resolver import resolve_objective_districts
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _market_metrics_key(
|
||
db: Session,
|
||
*,
|
||
district: str | None = None,
|
||
obj_ids: Sequence[int] | None = None,
|
||
window_months: int = 6,
|
||
premise_kind: str = "квартира",
|
||
obj_class: str | None = None,
|
||
room_bucket: str | None = None,
|
||
) -> tuple[Any, ...]:
|
||
"""Ключ кэша §22-форсайта для compute_market_metrics (#1129).
|
||
|
||
Все входы, влияющие на SQL-фильтр и метрики: district, набор obj_ids (tuple для
|
||
hashability), окно, premise_kind, + сегмент-оси obj_class/room_bucket (#1959 —
|
||
посегментный дефицит). В форсайт-пути obj_ids всегда None → tuple() устойчив.
|
||
`db` не в ключе (одна сессия на отчёт).
|
||
"""
|
||
return (
|
||
district,
|
||
tuple(obj_ids) if obj_ids is not None else None,
|
||
window_months,
|
||
premise_kind,
|
||
obj_class,
|
||
room_bucket,
|
||
)
|
||
|
||
|
||
Confidence = Literal["high", "medium", "low"]
|
||
|
||
# Лот считается «зависшим» (overstock), если он в продаже дольше этого числа
|
||
# месяцев и до сих пор не продан. ЕКБ-эмпирика: здоровый цикл поглощения ~ 12 мес.
|
||
_OVERSTOCK_MONTHS_THRESHOLD: int = 12
|
||
|
||
# Пороги уверенности по размеру выборки (зеркало духа ТЗ §15: мало лотов / 1 ЖК → low).
|
||
_CONF_HIGH_MIN_LOTS: int = 200
|
||
_CONF_HIGH_MIN_OBJ: int = 3
|
||
_CONF_MEDIUM_MIN_LOTS: int = 50
|
||
_CONF_MEDIUM_MIN_OBJ: int = 2
|
||
|
||
# Регион данных Объектива — ЕКБ (Свердловская обл.). Передаётся в elasticity-reuse,
|
||
# где параметр сохранён для обратной совместимости (objective покрывает только ЕКБ).
|
||
_EKB_REGION_CODE: int = 66
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MarketMetrics:
|
||
"""Рыночные метрики ТЗ §9.2 для локации (район и/или набор obj_ids).
|
||
|
||
Все метрики — детерминированные. Любая метрика = None при недостатке данных
|
||
(никогда 0-как-заглушка и никогда исключение).
|
||
"""
|
||
|
||
# ── Контекст выборки ──────────────────────────────────────────────────────
|
||
district: str | None
|
||
obj_count: int # сколько отдельных ЖК (project_name) попало в выборку
|
||
n_lots: int # всего лотов (квартир) в выборке
|
||
n_sold: int # из них проданных
|
||
n_available: int # из них доступных (в продаже)
|
||
window_months: int
|
||
premise_kind: str
|
||
confidence: Confidence
|
||
|
||
# ── §9.2 named-метрики ────────────────────────────────────────────────────
|
||
absorption_rate: float | None # ед./мес ÷ доступные ед. (доля поглощения в мес)
|
||
months_of_supply: float | None # доступные ед. ÷ месячное поглощение (мес до распродажи)
|
||
sell_through_pct: float | None # проданные ÷ (проданные + доступные), %
|
||
unit_velocity: float | None # ед. продано в месяц (за window_months)
|
||
area_velocity: float | None # м² продано в месяц (за window_months)
|
||
liquidity_index: dict[str, float] | None # {rooms_bucket: индекс относит. скорости}
|
||
overstock_index: float | None # доля долго-экспонируемого непроданного стока
|
||
demand_concentration: float | None # Херфиндаль продаж по комнатности (0..1)
|
||
price_sensitivity: float | None # эластичность цена↔спрос (slope, обычно < 0)
|
||
price_sensitivity_source: str | None # 'regression' | 'fallback' | None
|
||
# #1593: ед./мес per room-bucket (ключи — вокабуляр _room_bucket():
|
||
# "студия","1","2","3","4","5+"). None если нет данных в окне.
|
||
velocity_by_room: dict[str, float] | None
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"district": self.district,
|
||
"obj_count": self.obj_count,
|
||
"n_lots": self.n_lots,
|
||
"n_sold": self.n_sold,
|
||
"n_available": self.n_available,
|
||
"window_months": self.window_months,
|
||
"premise_kind": self.premise_kind,
|
||
"confidence": self.confidence,
|
||
"absorption_rate": _round_or_none(self.absorption_rate, 4),
|
||
"months_of_supply": _round_or_none(self.months_of_supply, 1),
|
||
"sell_through_pct": _round_or_none(self.sell_through_pct, 1),
|
||
"unit_velocity": _round_or_none(self.unit_velocity, 2),
|
||
"area_velocity": _round_or_none(self.area_velocity, 1),
|
||
"liquidity_index": (
|
||
{k: round(v, 3) for k, v in self.liquidity_index.items()}
|
||
if self.liquidity_index is not None
|
||
else None
|
||
),
|
||
"overstock_index": _round_or_none(self.overstock_index, 3),
|
||
"demand_concentration": _round_or_none(self.demand_concentration, 3),
|
||
"price_sensitivity": _round_or_none(self.price_sensitivity, 4),
|
||
"price_sensitivity_source": self.price_sensitivity_source,
|
||
# velocity_by_room намеренно не сериализуется: это внутренний
|
||
# pipeline-атрибут (потребляется recommendation.py напрямую).
|
||
}
|
||
|
||
|
||
def _round_or_none(value: float | None, digits: int) -> float | None:
|
||
return round(value, digits) if value is not None else None
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure-арифметика метрик — без БД, полностью юнит-тестируемо.
|
||
# Каждая функция graceful: пустой/нулевой вход → None (не 0, не ZeroDivisionError).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _monthly_rate(count: float | int | None, months: int) -> float | None:
|
||
"""count за окно → count в месяц. None/нет окна → None."""
|
||
if count is None or months <= 0:
|
||
return None
|
||
return float(count) / float(months)
|
||
|
||
|
||
def _absorption_rate(sold: int | None, available: int | None, months: int) -> float | None:
|
||
"""absorption_rate = (проданных в месяц) / доступных.
|
||
|
||
Доля текущего стока, поглощаемая рынком за месяц. Если нет доступных лотов
|
||
или нет окна — None (распродано / неизмеримо, НЕ 0).
|
||
"""
|
||
monthly_sold = _monthly_rate(sold, months)
|
||
if monthly_sold is None or not available or available <= 0:
|
||
return None
|
||
return monthly_sold / float(available)
|
||
|
||
|
||
def _months_of_supply(available: int | None, sold: int | None, months: int) -> float | None:
|
||
"""months_of_supply = доступные / (проданных в месяц).
|
||
|
||
Сколько месяцев нужно, чтобы распродать текущий сток при текущем темпе.
|
||
Нет продаж за окно (темп 0) → None (бесконечность неизмерима, НЕ 0).
|
||
"""
|
||
monthly_sold = _monthly_rate(sold, months)
|
||
if available is None or monthly_sold is None or monthly_sold <= 0:
|
||
return None
|
||
return float(available) / monthly_sold
|
||
|
||
|
||
def _sell_through_pct(sold: int | None, available: int | None) -> float | None:
|
||
"""sell_through_pct = sold / (sold + available) * 100.
|
||
|
||
Доля реализованного от всего выведенного на рынок. Пустая выборка → None.
|
||
"""
|
||
if sold is None or available is None:
|
||
return None
|
||
denom = sold + available
|
||
if denom <= 0:
|
||
return None
|
||
return float(sold) / float(denom) * 100.0
|
||
|
||
|
||
def _liquidity_index(sold_by_room: Mapping[str, int]) -> dict[str, float] | None:
|
||
"""liquidity_index per комнатность — относительная скорость продаж.
|
||
|
||
Нормируем долю продаж бакета на среднюю долю (1/n_buckets): индекс 1.0 =
|
||
«продаётся со средней по выборке скоростью», >1 = быстрее, <1 = медленнее.
|
||
Нет продаж ни в одном бакете → None.
|
||
"""
|
||
buckets = {k: int(v) for k, v in sold_by_room.items() if v is not None}
|
||
total = sum(buckets.values())
|
||
n = len(buckets)
|
||
if total <= 0 or n == 0:
|
||
return None
|
||
fair_share = 1.0 / n
|
||
return {bucket: (cnt / total) / fair_share for bucket, cnt in buckets.items()}
|
||
|
||
|
||
def _overstock_index(n_long_unsold: int | None, n_available: int | None) -> float | None:
|
||
"""overstock_index = долго-экспонируемые непроданные / все доступные.
|
||
|
||
Доля «зависшего» стока (в продаже > N месяцев без сделки). Нет доступных
|
||
лотов → None (неизмеримо, НЕ 0).
|
||
"""
|
||
if n_long_unsold is None or not n_available or n_available <= 0:
|
||
return None
|
||
return float(n_long_unsold) / float(n_available)
|
||
|
||
|
||
def _demand_concentration(sold_by_room: Mapping[str, int]) -> float | None:
|
||
"""demand_concentration — индекс Херфиндаля (HHI) долей продаж по комнатности.
|
||
|
||
Sum( share_i^2 ) ∈ (0..1]: 1.0 = весь спрос в одном формате, → 0 =
|
||
равномерно размазан. Нет продаж → None.
|
||
"""
|
||
counts = [int(v) for v in sold_by_room.values() if v is not None and v > 0]
|
||
total = sum(counts)
|
||
if total <= 0:
|
||
return None
|
||
return sum((c / total) ** 2 for c in counts)
|
||
|
||
|
||
def _confidence(n_lots: int, obj_count: int, n_sold: int) -> Confidence:
|
||
"""Уверенность по размеру выборки (ТЗ §15 spirit).
|
||
|
||
'low' если мало лотов / 1 ЖК / нет проданной истории — тогда метрики
|
||
скорости/поглощения статистически ненадёжны.
|
||
"""
|
||
if n_sold <= 0:
|
||
return "low"
|
||
if n_lots >= _CONF_HIGH_MIN_LOTS and obj_count >= _CONF_HIGH_MIN_OBJ:
|
||
return "high"
|
||
if n_lots >= _CONF_MEDIUM_MIN_LOTS and obj_count >= _CONF_MEDIUM_MIN_OBJ:
|
||
return "medium"
|
||
return "low"
|
||
|
||
|
||
def _room_bucket(rooms_int: int | None) -> str:
|
||
"""Нормализуем rooms_int (objective: 0=студия) в стабильный bucket-ключ."""
|
||
if rooms_int is None:
|
||
return "unknown"
|
||
if rooms_int <= 0:
|
||
return "студия"
|
||
if rooms_int >= 5:
|
||
return "5+"
|
||
return str(rooms_int)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# SQL aggregation
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
# Текущий сток per-flat. ДЕДУП до ПОСЛЕДНЕГО снапшота на физический лот (#1959):
|
||
# objective_lots — append-per-snapshot (~2.9× инфляция строк), без дедупа сток
|
||
# раздут и симметрично раздувает спрос И предложение → деградирует посегментный
|
||
# дефицит. Ключ физлота = (project_name, corpus_name, section, floor, lot_number);
|
||
# берём строку с MAX(snapshot_date), tie-break MAX(id). Поддерживается индексом
|
||
# objective_lots_physflat_latest_idx (создаётся в миграции 173) — он совпадает с
|
||
# ключом и направлениями этого DISTINCT ON → Index Only Scan + Unique без Sort.
|
||
# project_name NOT NULL; corpus/section/floor/lot_number могут быть NULL (~2.5%
|
||
# лотов) — для таких лотов DISTINCT ON схлопывает NULL=NULL в одну группу
|
||
# (неидентифицируемые физлоты, в дефицит-расчёте безопасно), для остального —
|
||
# точный per-flat дедуп.
|
||
#
|
||
# ⚠️ ДЕДУП БЕЗУСЛОВНЫЙ — применяется ко ВСЕМ вызовам compute_market_metrics, не
|
||
# только к форсайт-пути. Deep-review (#1959) подтвердил: безопасно для остальных
|
||
# потребителей (absorption/MoS/sell_through — отношения; velocity — saturated),
|
||
# т.к. inflation симметрична в числителе/знаменателе и съедается шкалой. Дедуп
|
||
# здесь касается ТОЛЬКО выборки внутри метрик; #1964 ввёл общий view
|
||
# v_objective_lots_latest для ОСТАЛЬНЫХ current-state консьюмеров — он использует
|
||
# ТОТ ЖЕ physflat-ключ/направления (project,corpus,section,floor,lot_number;
|
||
# snapshot_date DESC, id DESC), что и этот inline DISTINCT ON (инвариант общий).
|
||
# Этот блок НЕ рефакторим на view (zero behavior change, лишний риск).
|
||
#
|
||
# Сегмент-фильтры (#1959): obj_class (class в objective_lots — lowercase, матчим
|
||
# LOWER=LOWER) + room_bucket. ROOM-bucket — Source-B room_area-вокабуляр
|
||
# ("Студии 15-30"/"1-к 30-45"/"2-к 45-60"/"3-к 60-80"/"80+ м²"), ТОЧНОЕ зеркало
|
||
# sales_series.room_area_bucket_of (и _SOURCE_B_SQL): площадь ≥ :large_area → LARGE,
|
||
# иначе по rooms_int; так what_to_build (Source-B сетка) фильтрует БЕЗ перевода
|
||
# вокабуляра. Применяются ДО DISTINCT ON: сегментация объективна по физлоту
|
||
# (класс/комнатность/площадь не меняются между снапшотами).
|
||
#
|
||
# is_sold распознаём И через флаг is_sold, И через наличие contract_date / статус
|
||
# 'продан' — Объектив заполняет их неконсистентно. n_long_unsold: непродан и
|
||
# в продаже > N мес (sales_start_date — самый надёжный «когда вышел на рынок»).
|
||
_STOCK_SQL = text(
|
||
"""
|
||
WITH lots AS (
|
||
SELECT DISTINCT ON (
|
||
ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number
|
||
)
|
||
ol.project_name,
|
||
ol.rooms_int,
|
||
ol.area_pd,
|
||
ol.sales_start_date,
|
||
(
|
||
ol.is_sold IS TRUE
|
||
OR ol.contract_date IS NOT NULL
|
||
OR LOWER(COALESCE(ol.status, '')) = 'продан'
|
||
) AS sold_now
|
||
FROM objective_lots ol
|
||
WHERE ol.premise_kind = :premise_kind
|
||
AND (
|
||
CAST(:has_district AS boolean) IS FALSE
|
||
OR ol.district = ANY(CAST(:districts AS text[]))
|
||
)
|
||
AND (
|
||
CAST(:has_obj_ids AS boolean) IS FALSE
|
||
OR ol.objective_lot_id = ANY(CAST(:obj_ids AS bigint[]))
|
||
)
|
||
AND (
|
||
CAST(:cls AS text) IS NULL
|
||
OR LOWER(ol.class) = LOWER(CAST(:cls AS text))
|
||
)
|
||
AND (
|
||
CAST(:room_bucket AS text) IS NULL
|
||
OR (CASE
|
||
WHEN ol.area_pd >= CAST(:large_area AS numeric)
|
||
THEN CAST(:b_large AS text)
|
||
WHEN ol.rooms_int IS NULL THEN CAST(:b_unknown AS text)
|
||
WHEN ol.rooms_int <= 0 THEN CAST(:b_studio AS text)
|
||
WHEN ol.rooms_int = 1 THEN CAST(:b_1k AS text)
|
||
WHEN ol.rooms_int = 2 THEN CAST(:b_2k AS text)
|
||
WHEN ol.rooms_int = 3 THEN CAST(:b_3k AS text)
|
||
ELSE CAST(:b_large AS text)
|
||
END) = CAST(:room_bucket AS text)
|
||
)
|
||
ORDER BY
|
||
ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number,
|
||
ol.snapshot_date DESC, ol.id DESC
|
||
)
|
||
SELECT
|
||
COUNT(*) AS n_lots,
|
||
COUNT(*) FILTER (WHERE sold_now) AS n_sold,
|
||
COUNT(*) FILTER (WHERE NOT sold_now) AS n_available,
|
||
COUNT(DISTINCT project_name) AS obj_count,
|
||
COUNT(*) FILTER (
|
||
WHERE NOT sold_now
|
||
AND sales_start_date IS NOT NULL
|
||
AND sales_start_date
|
||
<= CURRENT_DATE - CAST(:overstock_interval AS interval)
|
||
) AS n_long_unsold
|
||
FROM lots
|
||
"""
|
||
)
|
||
|
||
# Продажи за окно — напрямую из objective_lots по contract_date («Дата договора»).
|
||
# «Продано в окне» = лот с contract_date внутри окна. contract_date — реальная дата
|
||
# сделки и заполнен у 100% проданных лотов, поэтому источник надёжен и не требует
|
||
# истории. (Раньше считали через objective_lots_history-снапшоты, но history глубиной
|
||
# ~17 дней: любой сейчас-проданный лот имел sold-снапшот в окне → «продажи в окне»
|
||
# схлопывались в весь кумулятивный проданный сток, завышая absorption/velocity/MoS.
|
||
# Bug #949: Автовокзал 6mo давал ~33 245 ед. вместо реальных ~2 308.)
|
||
#
|
||
# ДЕДУП до ПОСЛЕДНЕГО снапшота на физический лот (#1959, симметрично _STOCK_SQL):
|
||
# без дедупа один проданный физлот, присутствующий в N снапшотах, считается N раз
|
||
# → velocity/absorption раздуты ~2.9× (раздувают спрос симметрично раздутому
|
||
# предложению → дефицит вырождается в −1.0). Дедупим в CTE `latest`, окно по
|
||
# contract_date и сегмент-фильтры применяем уже к дедуплицированному набору.
|
||
# Ключ/направления дедупа — зеркало _STOCK_SQL (оба поддержаны индексом
|
||
# objective_lots_physflat_latest_idx из миграции 173). contract_date / rooms_int /
|
||
# area_pd берём из последнего снапшота физлота.
|
||
# Сегмент-фильтры (#1959): obj_class (lowercase, LOWER=LOWER) + room_bucket
|
||
# (Source-B room_area-вокабуляр "Студии 15-30"/"1-к 30-45"/"2-к 45-60"/"3-к 60-80"/
|
||
# "80+ м²" — ТОЧНОЕ зеркало sales_series.room_area_bucket_of, как в _STOCK_SQL) —
|
||
# применяются ПОСЛЕ дедупа (объективны по физлоту: класс/комнатность/площадь не
|
||
# меняются между снапшотами). ROLLUP остаётся: при room_bucket-фильтре отдаёт
|
||
# {bucket, total}.
|
||
_SALES_WINDOW_SQL = text(
|
||
"""
|
||
WITH latest AS (
|
||
SELECT DISTINCT ON (
|
||
ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number
|
||
)
|
||
ol.rooms_int,
|
||
ol.area_pd,
|
||
ol.contract_date,
|
||
ol.class
|
||
FROM objective_lots ol
|
||
WHERE ol.premise_kind = :premise_kind
|
||
AND (
|
||
CAST(:has_district AS boolean) IS FALSE
|
||
OR ol.district = ANY(CAST(:districts AS text[]))
|
||
)
|
||
AND (
|
||
CAST(:has_obj_ids AS boolean) IS FALSE
|
||
OR ol.objective_lot_id = ANY(CAST(:obj_ids AS bigint[]))
|
||
)
|
||
ORDER BY
|
||
ol.project_name, ol.corpus_name, ol.section, ol.floor, ol.lot_number,
|
||
ol.snapshot_date DESC, ol.id DESC
|
||
)
|
||
SELECT
|
||
COUNT(*) AS units_sold_window,
|
||
COALESCE(SUM(area_pd), 0) AS area_sold_window,
|
||
rooms_int,
|
||
-- #1214: ROLLUP grand-total и NULL-группа дают rooms_int IS NULL обе.
|
||
-- Различаем через GROUPING(): =1 для grand-total, =0 для NULL-комнатной
|
||
-- группы. Без этого один проданный лот с rooms_int IS NULL даёт ДВЕ
|
||
-- строки rooms_int IS NULL (NULL-группа + итог), и MixedAggregate-план
|
||
-- эмитит итог ПЕРВЫМ → NULL-группа затирает units_total частичным
|
||
-- счётом → unit_velocity/absorption занижены, MoS завышен.
|
||
GROUPING(rooms_int) AS is_total
|
||
FROM latest
|
||
WHERE contract_date IS NOT NULL
|
||
AND contract_date >= CURRENT_DATE - CAST(:window_interval AS interval)
|
||
AND (
|
||
CAST(:cls AS text) IS NULL
|
||
OR LOWER(class) = LOWER(CAST(:cls AS text))
|
||
)
|
||
AND (
|
||
CAST(:room_bucket AS text) IS NULL
|
||
OR (CASE
|
||
WHEN area_pd >= CAST(:large_area AS numeric)
|
||
THEN CAST(:b_large AS text)
|
||
WHEN rooms_int IS NULL THEN CAST(:b_unknown AS text)
|
||
WHEN rooms_int <= 0 THEN CAST(:b_studio AS text)
|
||
WHEN rooms_int = 1 THEN CAST(:b_1k AS text)
|
||
WHEN rooms_int = 2 THEN CAST(:b_2k AS text)
|
||
WHEN rooms_int = 3 THEN CAST(:b_3k AS text)
|
||
ELSE CAST(:b_large AS text)
|
||
END) = CAST(:room_bucket AS text)
|
||
)
|
||
GROUP BY ROLLUP (rooms_int)
|
||
"""
|
||
)
|
||
|
||
|
||
@cached(_market_metrics_key, label="compute_market_metrics")
|
||
def compute_market_metrics(
|
||
db: Session,
|
||
*,
|
||
district: str | None = None,
|
||
obj_ids: Sequence[int] | None = None,
|
||
window_months: int = 6,
|
||
premise_kind: str = "квартира",
|
||
obj_class: str | None = None,
|
||
room_bucket: str | None = None,
|
||
) -> MarketMetrics:
|
||
"""Вычислить рыночные метрики ТЗ §9.2 для локации.
|
||
|
||
Фильтрация по `district` и/или `obj_ids` (объединяются по AND, если оба
|
||
заданы). Если ни один не задан — считается по всей выборке premise_kind
|
||
(имеет смысл для ЕКБ-wide baseline).
|
||
|
||
#1959 (посегментный дефицит): `obj_class` (lowercase в objective_lots,
|
||
матчим регистронезависимо) и `room_bucket` (Source-B room_area-вокабуляр —
|
||
"Студии 15-30"/"1-к 30-45"/"2-к 45-60"/"3-к 60-80"/"80+ м²", зеркало
|
||
`sales_series.room_area_bucket_of` → what_to_build фильтрует БЕЗ перевода)
|
||
сужают выборку до одного сегмента. Так base_pace (unit_velocity) и доступный
|
||
сток становятся ПОСЕГМЕНТНЫМИ, а не district-wide (иначе все ячейки
|
||
what_to_build делят один темп → дефицит вырождается в −1.0). None по обеим осям
|
||
→ прежнее district-wide поведение. ВАЖНО: SQL дедуплицирует objective_lots до
|
||
последнего снапшота на физлот → counts (а значит и confidence) считаются на
|
||
дедуплицированном наборе.
|
||
|
||
Возвращает MarketMetrics ВСЕГДА (даже на пустых данных): тогда метрики =
|
||
None, confidence='low'. Никогда не бросает на отсутствии данных.
|
||
"""
|
||
# #1959 room-bucket пороги/метки — зеркало sales_series.room_area_bucket_of /
|
||
# _SOURCE_B_SQL (Source-B вокабуляр), чтобы what_to_build фильтровал без перевода.
|
||
# Локальный (lazy) импорт: модульный импорт sales_series тянет forecasting/__init__
|
||
# → demand_supply_forecast → future_supply → market_metrics (циклический импорт).
|
||
from app.services.forecasting.sales_series import (
|
||
_LARGE_AREA_THRESHOLD_M2,
|
||
ROOM_AREA_BUCKET_1K,
|
||
ROOM_AREA_BUCKET_2K,
|
||
ROOM_AREA_BUCKET_3K,
|
||
ROOM_AREA_BUCKET_LARGE,
|
||
ROOM_AREA_BUCKET_STUDIO,
|
||
ROOM_AREA_BUCKET_UNKNOWN,
|
||
)
|
||
|
||
obj_id_list: list[int] = [int(x) for x in obj_ids] if obj_ids else []
|
||
has_obj_ids = bool(obj_id_list)
|
||
# Резолвим district (админ-имя ЕКБ) → набор informal микро-районов, по которым
|
||
# реально фильтруется objective_lots. None → EKB-wide (без district-фильтра); это
|
||
# лечит баг «админ-имя → 0 строк → пустой прогноз» (см. district_resolver).
|
||
micros = resolve_objective_districts(db, district)
|
||
has_district = micros is not None
|
||
params: dict[str, Any] = {
|
||
"premise_kind": premise_kind,
|
||
# has_district=False → district-ветка фильтра отключена (EKB-wide).
|
||
"has_district": has_district,
|
||
# ANY(NULL::text[]) валиден; пустой список когда фильтра нет (has_district=False).
|
||
"districts": micros if micros is not None else [],
|
||
"has_obj_ids": has_obj_ids,
|
||
# ANY(NULL::bigint[]) валиден; передаём пустой список когда фильтра нет.
|
||
"obj_ids": obj_id_list,
|
||
"overstock_interval": f"{_OVERSTOCK_MONTHS_THRESHOLD} months",
|
||
# #1959 сегмент-фильтры (None → ось не сужается).
|
||
"cls": obj_class,
|
||
"room_bucket": room_bucket,
|
||
# #1959 room-bucket пороги/метки — зеркало sales_series.room_area_bucket_of
|
||
# / _SOURCE_B_SQL (Source-B вокабуляр), чтобы what_to_build фильтровал без
|
||
# перевода. Меняешь bucketing — правь в ОБОИХ местах (тут + sales_series).
|
||
"large_area": _LARGE_AREA_THRESHOLD_M2,
|
||
"b_studio": ROOM_AREA_BUCKET_STUDIO,
|
||
"b_1k": ROOM_AREA_BUCKET_1K,
|
||
"b_2k": ROOM_AREA_BUCKET_2K,
|
||
"b_3k": ROOM_AREA_BUCKET_3K,
|
||
"b_large": ROOM_AREA_BUCKET_LARGE,
|
||
"b_unknown": ROOM_AREA_BUCKET_UNKNOWN,
|
||
}
|
||
|
||
# ── Текущий сток ──────────────────────────────────────────────────────────
|
||
stock = _query_stock(db, params)
|
||
n_lots = stock["n_lots"]
|
||
n_sold_total = stock["n_sold"]
|
||
n_available = stock["n_available"]
|
||
obj_count = stock["obj_count"]
|
||
n_long_unsold = stock["n_long_unsold"]
|
||
|
||
# ── Продажи за окно (для velocity / absorption / liquidity / concentration) ─
|
||
window_params = {**params, "window_interval": f"{window_months} months"}
|
||
units_sold_window, area_sold_window, sold_by_room = _query_sales_window(db, window_params)
|
||
|
||
# ── Pure-метрики ──────────────────────────────────────────────────────────
|
||
# n_lots == 0 → выборка пуста, мерить нечего: velocity/absorption = None
|
||
# (НЕ 0 — иначе «нет данных» не отличить от «честно продали 0»). При n_lots>0
|
||
# и нуле продаж в окне velocity=0.0 — это валидное измерение «0 ед./мес».
|
||
has_sample = n_lots > 0
|
||
units_window: int | None = units_sold_window if has_sample else None
|
||
area_window: float | None = area_sold_window if has_sample else None
|
||
absorption = _absorption_rate(units_window, n_available, window_months)
|
||
mos = _months_of_supply(n_available, units_window, window_months)
|
||
sell_through = _sell_through_pct(n_sold_total, n_available)
|
||
unit_velocity = _monthly_rate(units_window, window_months)
|
||
area_velocity = _monthly_rate(area_window, window_months)
|
||
liquidity = _liquidity_index(sold_by_room)
|
||
overstock = _overstock_index(n_long_unsold, n_available)
|
||
demand_conc = _demand_concentration(sold_by_room)
|
||
# #1593: per-bucket velocity — ед./мес по каждой комнатности. Ключи зеркалят
|
||
# _room_bucket() ("студия","1","2","3","4","5+"). При has_sample=False нет
|
||
# смысла делить 0 лотов → None (graceful, зеркало unit_velocity поведения).
|
||
# window_months > 0 защищает от ZeroDivisionError (зеркало _monthly_rate()).
|
||
vel_by_room: dict[str, float] | None = (
|
||
{bkt: float(cnt) / float(window_months) for bkt, cnt in sold_by_room.items()}
|
||
if has_sample and sold_by_room and window_months > 0
|
||
else None
|
||
)
|
||
|
||
# ── price_sensitivity — reuse analytics_queries._elasticity_coef ───────────
|
||
price_sensitivity, price_sensitivity_source = _price_sensitivity(
|
||
db, district=district, window_months=window_months
|
||
)
|
||
|
||
confidence = _confidence(n_lots=n_lots, obj_count=obj_count, n_sold=n_sold_total)
|
||
|
||
logger.info(
|
||
"market_metrics: district=%s micros=%s class=%s room=%s obj_ids=%d n_lots=%d "
|
||
"n_sold=%d n_available=%d obj_count=%d units_sold_window=%d confidence=%s",
|
||
district,
|
||
micros,
|
||
obj_class,
|
||
room_bucket,
|
||
len(obj_id_list),
|
||
n_lots,
|
||
n_sold_total,
|
||
n_available,
|
||
obj_count,
|
||
units_sold_window,
|
||
confidence,
|
||
)
|
||
|
||
return MarketMetrics(
|
||
district=district,
|
||
obj_count=obj_count,
|
||
n_lots=n_lots,
|
||
n_sold=n_sold_total,
|
||
n_available=n_available,
|
||
window_months=window_months,
|
||
premise_kind=premise_kind,
|
||
confidence=confidence,
|
||
absorption_rate=absorption,
|
||
months_of_supply=mos,
|
||
sell_through_pct=sell_through,
|
||
unit_velocity=unit_velocity,
|
||
area_velocity=area_velocity,
|
||
liquidity_index=liquidity,
|
||
overstock_index=overstock,
|
||
demand_concentration=demand_conc,
|
||
price_sensitivity=price_sensitivity,
|
||
price_sensitivity_source=price_sensitivity_source,
|
||
velocity_by_room=vel_by_room,
|
||
)
|
||
|
||
|
||
def _query_stock(db: Session, params: Mapping[str, Any]) -> dict[str, int]:
|
||
"""Текущий сток. На ошибке/пустых данных → все счётчики 0 (graceful)."""
|
||
try:
|
||
row = db.execute(_STOCK_SQL, dict(params)).mappings().first()
|
||
except Exception:
|
||
logger.exception(
|
||
"market_metrics: stock query failed (districts=%s)", params.get("districts")
|
||
)
|
||
row = None
|
||
if row is None:
|
||
return {
|
||
"n_lots": 0,
|
||
"n_sold": 0,
|
||
"n_available": 0,
|
||
"obj_count": 0,
|
||
"n_long_unsold": 0,
|
||
}
|
||
return {
|
||
"n_lots": int(row["n_lots"] or 0),
|
||
"n_sold": int(row["n_sold"] or 0),
|
||
"n_available": int(row["n_available"] or 0),
|
||
"obj_count": int(row["obj_count"] or 0),
|
||
"n_long_unsold": int(row["n_long_unsold"] or 0),
|
||
}
|
||
|
||
|
||
def _query_sales_window(
|
||
db: Session, params: Mapping[str, Any]
|
||
) -> tuple[int, float, dict[str, int]]:
|
||
"""Продажи за окно по contract_date. Возвращает (units, area_m2, {bucket: units}).
|
||
|
||
GROUP BY ROLLUP с GROUPING(rooms_int) AS is_total (#1214):
|
||
• is_total=1 → grand-total (units/area за все комнаты);
|
||
• is_total=0 и rooms_int IS NULL → разбивка для лотов БЕЗ rooms — кладём
|
||
в by_room['unknown'] (а не путаем с total);
|
||
• is_total=0 и rooms_int не NULL → разбивка по конкретной комнатности.
|
||
by_room аккумулирует через += чтобы при будущих доп.NULL-вариантах не
|
||
затирать прежние счётчики. На ошибке/пусто → (0, 0.0, {}).
|
||
"""
|
||
try:
|
||
rows = db.execute(_SALES_WINDOW_SQL, dict(params)).mappings().all()
|
||
except Exception:
|
||
logger.exception(
|
||
"market_metrics: sales-window query failed (districts=%s)", params.get("districts")
|
||
)
|
||
rows = []
|
||
|
||
units_total = 0
|
||
area_total = 0.0
|
||
by_room: dict[str, int] = {}
|
||
for r in rows:
|
||
cnt = int(r["units_sold_window"] or 0)
|
||
area = float(r["area_sold_window"] or 0.0)
|
||
if int(r["is_total"]) == 1:
|
||
# ROLLUP grand-total — единственная строка с GROUPING=1.
|
||
units_total = cnt
|
||
area_total = area
|
||
elif r["rooms_int"] is None:
|
||
# Лоты с rooms_int IS NULL (ETL пишет NULL для «неопределённого типа»)
|
||
# — отдельный бакет, не путаем с total.
|
||
by_room["unknown"] = by_room.get("unknown", 0) + cnt
|
||
else:
|
||
bucket = _room_bucket(int(r["rooms_int"]))
|
||
by_room[bucket] = by_room.get(bucket, 0) + cnt
|
||
return units_total, area_total, by_room
|
||
|
||
|
||
def _price_sensitivity(
|
||
db: Session, *, district: str | None, window_months: int
|
||
) -> tuple[float | None, str | None]:
|
||
"""Эластичность цена↔спрос — reuse analytics_queries._elasticity_coef.
|
||
|
||
Требует district (регрессия по району). Без district → None (нечего фитить).
|
||
elasticity-окно отдельно от velocity-окна: регрессии нужно больше истории,
|
||
поэтому минимум 24 мес (как в recommend_mix).
|
||
|
||
#1211 fix: District-вход — admin-имя ЕКБ (из /analyze), но
|
||
``objective_corpus_room_month.district`` — МИКРО-вокабуляр (как и
|
||
``objective_lots.district``). Сырое admin-имя в фильтр давало 0 точек
|
||
регрессии → всегда FALLBACK_ELASTICITY (silent-correctness). Резолвим
|
||
admin → набор микро через ``resolve_objective_districts`` и передаём
|
||
списком в ``_elasticity_coef(..., districts=...)``. Резолвер None
|
||
(нет чистых алиасов / 'не определён') → пустой список → EKB-wide
|
||
регрессия (лучше, чем 0 точек: получаем агрегированный сигнал).
|
||
"""
|
||
if not district:
|
||
return None, None
|
||
micros = resolve_objective_districts(db, district)
|
||
elasticity_window = max(window_months, 24)
|
||
try:
|
||
elast = _elasticity_coef(
|
||
db,
|
||
region_code=_EKB_REGION_CODE,
|
||
district_name=district,
|
||
target_class=None,
|
||
elasticity_window_months=elasticity_window,
|
||
districts=micros if micros is not None else [],
|
||
)
|
||
except Exception:
|
||
logger.exception(
|
||
"market_metrics: elasticity reuse failed (district=%s micros=%s)",
|
||
district,
|
||
micros,
|
||
)
|
||
return None, None
|
||
return float(elast["elasticity"]), str(elast["source"])
|