Light1YT
c4a06813c2
Deterministic, no-LLM market metrics from Objective data for a location (district and/or obj_ids): absorption_rate, months_of_supply, sell_through_pct, unit/area_velocity, liquidity_index per room-type, overstock_index, demand_concentration (HHI), price_sensitivity (reuses analytics_queries._elasticity_coef). Frozen MarketMetrics dataclass + as_dict, sample-size confidence (§15 spirit). Graceful on thin data (the #1 coverage risk): no data → metric None (never 0/crash), confidence='low'. Filters by district/obj_ids to sidestep the sparse domrf↔objective mapping. Read-only SELECTs, 55 unit/mock tests. Consumed by forecast (#952) + relevance (#949 PR B). No migration, no endpoint.
506 lines
24 KiB
Python
506 lines
24 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).
|
||
- `objective_lots_history` — weekly-снапшоты per-flat: is_sold, contract_date,
|
||
area_pd — time-series для velocity/sell-through.
|
||
- 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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
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
|
||
|
||
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,
|
||
}
|
||
|
||
|
||
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. Считаем по objective_lots (последний UPSERT-снапшот).
|
||
# is_sold распознаём И через флаг is_sold, И через наличие contract_date / статус
|
||
# 'продан' — Объектив заполняет их неконсистентно. n_long_unsold: непродан и
|
||
# в продаже > N мес (sales_start_date — самый надёжный «когда вышел на рынок»).
|
||
_STOCK_SQL = text(
|
||
"""
|
||
WITH lots AS (
|
||
SELECT
|
||
ol.objective_lot_id,
|
||
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(:district AS text) IS NULL
|
||
OR ol.district = CAST(:district AS text)
|
||
)
|
||
AND (
|
||
CAST(:has_obj_ids AS boolean) IS FALSE
|
||
OR ol.objective_lot_id = ANY(CAST(:obj_ids AS bigint[]))
|
||
)
|
||
)
|
||
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_history (weekly timeline). «Продано в окне»
|
||
# = лот, у которого в окне появилась contract_date ИЛИ is_sold стал TRUE. Берём
|
||
# по одному событию на лот (MIN snapshot где он впервые помечен проданным),
|
||
# чтобы не считать один и тот же лот многократно из-за повторных снапшотов.
|
||
# area_pd берём из того же history-снапшота (последний known area для лота).
|
||
_SALES_WINDOW_SQL = text(
|
||
"""
|
||
WITH hist AS (
|
||
SELECT
|
||
h.objective_lot_id,
|
||
h.snapshot_date,
|
||
h.area_pd,
|
||
(h.is_sold IS TRUE OR h.contract_date IS NOT NULL) AS sold_flag
|
||
FROM objective_lots_history h
|
||
JOIN objective_lots ol
|
||
ON ol.objective_lot_id = h.objective_lot_id
|
||
WHERE ol.premise_kind = :premise_kind
|
||
AND (
|
||
CAST(:district AS text) IS NULL
|
||
OR ol.district = CAST(:district AS text)
|
||
)
|
||
AND (
|
||
CAST(:has_obj_ids AS boolean) IS FALSE
|
||
OR ol.objective_lot_id = ANY(CAST(:obj_ids AS bigint[]))
|
||
)
|
||
AND h.snapshot_date >= CURRENT_DATE - CAST(:window_interval AS interval)
|
||
),
|
||
first_sold AS (
|
||
SELECT DISTINCT ON (objective_lot_id)
|
||
objective_lot_id,
|
||
ol2.rooms_int,
|
||
hist.area_pd
|
||
FROM hist
|
||
JOIN objective_lots ol2 USING (objective_lot_id)
|
||
WHERE hist.sold_flag
|
||
ORDER BY objective_lot_id, hist.snapshot_date ASC
|
||
)
|
||
SELECT
|
||
COUNT(*) AS units_sold_window,
|
||
COALESCE(SUM(area_pd), 0) AS area_sold_window,
|
||
rooms_int
|
||
FROM first_sold
|
||
GROUP BY ROLLUP (rooms_int)
|
||
"""
|
||
)
|
||
|
||
|
||
def compute_market_metrics(
|
||
db: Session,
|
||
*,
|
||
district: str | None = None,
|
||
obj_ids: Sequence[int] | None = None,
|
||
window_months: int = 6,
|
||
premise_kind: str = "квартира",
|
||
) -> MarketMetrics:
|
||
"""Вычислить рыночные метрики ТЗ §9.2 для локации.
|
||
|
||
Фильтрация по `district` и/или `obj_ids` (объединяются по AND, если оба
|
||
заданы). Если ни один не задан — считается по всей выборке premise_kind
|
||
(имеет смысл для ЕКБ-wide baseline).
|
||
|
||
Возвращает MarketMetrics ВСЕГДА (даже на пустых данных): тогда метрики =
|
||
None, confidence='low'. Никогда не бросает на отсутствии данных.
|
||
"""
|
||
obj_id_list: list[int] = [int(x) for x in obj_ids] if obj_ids else []
|
||
has_obj_ids = bool(obj_id_list)
|
||
params: dict[str, Any] = {
|
||
"premise_kind": premise_kind,
|
||
"district": district,
|
||
"has_obj_ids": has_obj_ids,
|
||
# ANY(NULL::bigint[]) валиден; передаём пустой список когда фильтра нет.
|
||
"obj_ids": obj_id_list,
|
||
"overstock_interval": f"{_OVERSTOCK_MONTHS_THRESHOLD} months",
|
||
}
|
||
|
||
# ── Текущий сток ──────────────────────────────────────────────────────────
|
||
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)
|
||
|
||
# ── 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 obj_ids=%d n_lots=%d n_sold=%d n_available=%d "
|
||
"obj_count=%d units_sold_window=%d confidence=%s",
|
||
district,
|
||
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,
|
||
)
|
||
|
||
|
||
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 (district=%s)", params.get("district"))
|
||
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]]:
|
||
"""Продажи за окно из history. Возвращает (units, area_m2, {bucket: units}).
|
||
|
||
GROUP BY ROLLUP: строка с rooms_int IS NULL — это grand-total (берём как
|
||
units/area), остальные строки — разбивка по комнатности (для liquidity /
|
||
demand_concentration). На ошибке/пусто → (0, 0.0, {}).
|
||
"""
|
||
try:
|
||
rows = db.execute(_SALES_WINDOW_SQL, dict(params)).mappings().all()
|
||
except Exception:
|
||
logger.exception(
|
||
"market_metrics: sales-window query failed (district=%s)", params.get("district")
|
||
)
|
||
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 r["rooms_int"] is None:
|
||
# ROLLUP grand-total.
|
||
units_total = cnt
|
||
area_total = area
|
||
else:
|
||
by_room[_room_bucket(int(r["rooms_int"]))] = 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).
|
||
"""
|
||
if not district:
|
||
return None, None
|
||
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,
|
||
)
|
||
except Exception:
|
||
logger.exception("market_metrics: elasticity reuse failed (district=%s)", district)
|
||
return None, None
|
||
return float(elast["elasticity"]), str(elast["source"])
|