gendesign/backend/app/services/site_finder/market_metrics.py

"""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 = "квартира",
) -> tuple[Any, ...]:
    """Ключ кэша §22-форсайта для compute_market_metrics (#1129).

    Все входы, влияющие на SQL-фильтр и метрики: district, набор obj_ids (tuple для
    hashability), окно, premise_kind. В форсайт-пути obj_ids всегда None → tuple()
    устойчив. `db` не в ключе (одна сессия на отчёт).
    """
    return (
        district,
        tuple(obj_ids) if obj_ids is not None else None,
        window_months,
        premise_kind,
    )

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(: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[]))
               )
    )
    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.)
# area_pd берём из самого objective_lots (текущий per-flat area).
_SALES_WINDOW_SQL = text(
    """
    SELECT
        COUNT(*)                          AS units_sold_window,
        COALESCE(SUM(area_pd), 0)         AS area_sold_window,
        rooms_int
      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 ol.contract_date IS NOT NULL
       AND ol.contract_date >= CURRENT_DATE - CAST(:window_interval AS interval)
     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 = "квартира",
) -> 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)
    # Резолвим 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",
    }

    # ── Текущий сток ──────────────────────────────────────────────────────────
    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 micros=%s obj_ids=%d n_lots=%d n_sold=%d "
        "n_available=%d obj_count=%d units_sold_window=%d confidence=%s",
        district,
        micros,
        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 (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: строка с 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 (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 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"])