gendesign/tradein-mvp/backend/app/services/location_index.py

"""Location index for trade-in estimates — replaces the broken `location_coef` (LocationDrawer).

ИСТОРИЯ / ПОЧЕМУ ПЕРЕПИСАНО:
Старый `location_coef.py` считал `coef = 0.95 + (poi_weighted_score/100) * 0.10` — диапазон
жёстко зажат в [0.95, 1.05], без какой-либо калибровки на реальных ценах. Аудит на боевой БД
(1500 адресов ЕКБ + 4000 активных лотов) показал:
  - 67% адресов попадали в −1%…+1%, у ~25% coef был РОВНО 1.0 (score=50) — почти
    неинформативно, весь город умещался в −4%…+5%;
  - связи с ценой не было вообще: медиана ₽/м² по бакетам coef плоская и НЕ монотонна
    (бакет −4% дороже бакета +3%).
Для сравнения, расстояние до центра ЕКБ на 31 тыс. лотов даёт чистый монотонный градиент
(0-2км 249 686 ₽/м² → 12-13км 93 677 ₽/м², разброс 2.7×) — сигнал в данных есть, просто
POI-score его не улавливал (POI ranking ≠ цена).

НОВЫЙ ПОКАЗАТЕЛЬ (location index):
  location_index_pct = (медиана ₽/м² сопоставимых активных листингов в радиусе точки −
                         медиана ₽/м² по всему ЕКБ) / медиана по ЕКБ * 100

Самообновляем (те же `listings`, что уже скрейпятся под estimator), интерпретируем напрямую
("район на N% дороже/дешевле среднего по городу"), устойчив к выбросам (percentile_cont(0.5) —
медиана самой природой игнорирует единичные экстремумы, в отличие от mean/min/max), и НЕ зажат
искусственно — если район реально на 40% дороже, так и покажет.

ЧЕСТНАЯ ДЕГРАДАЦИЯ (см. LocationIndexResult.status):
  - "out_of_coverage"   — точка вне гео-охвата продукта (bbox Екатеринбурга). НЕ 0%, НЕ
    fallback-число — прочерк на фронте.
  - "insufficient_data" — даже на максимальном радиусе выборки < MIN_SAMPLE_SIZE сопоставимых
    активных листингов. Тоже прочерк, а не шум по 3 объявлениям.
  - "ok"                — index надёжен.

В ЦЕНУ НЕ ИДЁТ: estimator.py про этот модуль не знает и не должен знать — аналоги уже берутся
из того же района (локация учтена в базовой цене через сам подбор сопоставимых объектов),
повторное умножение на локационный индекс было бы двойным учётом одного и того же эффекта.

POI («что рядом» — школа/метро/остановка) сохранены как ОТДЕЛЬНАЯ качественная справка
(`nearby_poi`, ранжирование как раньше в location_coef.py), но больше не участвуют в числовом
показателе.
"""

from __future__ import annotations

import logging
from typing import Any

from pydantic import BaseModel
from sqlalchemy import text

logger = logging.getLogger(__name__)

# ── Гео-охват продукта: только Екатеринбург ──────────────────────────────────
# Тот же bbox, что EKB_BBOX в backend/app/services/site_finder/poi_loader.py (main
# gendesign backend, Overpass-загрузчик osm_poi_ekb) и что использовался при аудите
# (1500 адресов / 4000 активных лотов / 2787 POI, все — "по Екатеринбургу"). tradein-mvp —
# отдельный деплой/venv от backend/, поэтому константа продублирована, не импортирована;
# при изменении bbox в одном месте — проверить и второе (комментарий в обе стороны).
_EKB_BBOX_SOUTH = 56.70
_EKB_BBOX_WEST = 60.50
_EKB_BBOX_NORTH = 56.95
_EKB_BBOX_EAST = 60.75


def _in_ekb_bbox(lat: float, lon: float) -> bool:
    """True если точка внутри гео-охвата продукта (Екатеринбург)."""
    return _EKB_BBOX_SOUTH <= lat <= _EKB_BBOX_NORTH and _EKB_BBOX_WEST <= lon <= _EKB_BBOX_EAST


# ── Калибровочные константы (радиус / минимальная выборка) ──────────────────
#
# Плотность-прикидка для обоснования порядка величины (НЕ подтверждено живым запросом к
# прод-БД в этом изменении — см. PR description "непроверенное"): ЕКБ-аудит насчитал ~4000
# активных лотов в bbox площадью ~ 27.8км (0.25° широты) × 15.3км (0.25° долготы на широте
# 56.8°) ≈ 425 км² → плотность ~9.4 лота/км². Круг радиусом 800м имеет площадь ~2.01 км² →
# ожидаемо ~19 лотов при равномерной плотности — близко к MIN_SAMPLE_SIZE=20, т.е. стартовый
# радиус разумен для "средней" точки. Плотность в городе крайне неравномерна (центр много
# гуще окраин) — поэтому лестница радиусов расширяется, а не фиксированный радиус.
RADIUS_LADDER_M: tuple[int, ...] = (800, 1500, 2500)

# Ниже этого числа сопоставимых активных листингов медиана — шум, не показатель.
# Порог не откалиброван статистически (например через доверительный интервал медианы) —
# первая рабочая оценка для MVP. TODO: перепроверить на реальном распределении выборок по
# районам ЕКБ (см. "непроверенное" в отчёте задачи).
MIN_SAMPLE_SIZE = 20

# Санитарные (НЕ бизнес-калибровочные) границы ₽/м² — отсекают заведомо битые скрейп-строки
# (парсинг ошибся на порядок и т.п.), не сужают реальный рынок ЕКБ (там диапазон примерно
# 40-400 тыс₽/м², с большим запасом по краям).
_PRICE_PER_M2_SANITY_MIN = 20_000
_PRICE_PER_M2_SANITY_MAX = 1_000_000

DEFAULT_POI_RADIUS_M = 1200  # как в старом location_coef.py — подобран для МКД
DEFAULT_POI_TOP_N = 7

# Веса по категории POI — те же, что были в location_coef.py (ranking "что рядом",
# больше НЕ конвертируются в число, влияющее на индекс).
CATEGORY_WEIGHTS: dict[str, float] = {
    "metro_stop": 6.0,
    "school": 5.0,
    "kindergarten": 4.5,
    "hospital": 4.0,
    "shop_mall": 4.0,
    "shop_supermarket": 3.5,
    "bus_stop": 4.5,
    "park": 3.5,
    "pharmacy": 2.5,
    "tram_stop": 2.0,
    "shop_small": 2.0,
    "default": 1.0,
}


def _category_weight(category: str | None) -> float:
    """Вернуть вес категории. Если не знаем — default."""
    return CATEGORY_WEIGHTS.get(category or "default", CATEGORY_WEIGHTS["default"])


class NearbyPoi(BaseModel):
    """Один пункт «что рядом» — качественная справка, НЕ участвует в location_index_pct."""

    poi_type: str
    name: str | None
    distance_m: float


class LocationIndexResult(BaseModel):
    """Результат compute_location_index — потребляется эндпоинтом location-index."""

    status: str  # "ok" | "out_of_coverage" | "insufficient_data"
    location_index_pct: float | None
    local_median_price_per_m2: int | None
    city_median_price_per_m2: int | None
    sample_size: int
    radius_m: int
    nearby_poi: list[NearbyPoi]
    poi_status: str  # "ok" | "unavailable" (osm_poi_ekb_local пуста/не отрефрешена)


def _pct_deviation(local_median_ppm2: float, city_median_ppm2: float) -> float:
    """% отклонения локальной медианы от городской.

    Округление до 1 знака — не создаёт ложной точности (исходные данные — шумные скрейп-цены).
    """
    if city_median_ppm2 <= 0:
        # Защита от деления на ноль при вырожденной городской выборке — не должно
        # случаться в проде (там ~4000 активных лотов), только в пустой dev-БД.
        return 0.0
    return round((local_median_ppm2 - city_median_ppm2) / city_median_ppm2 * 100.0, 1)


# ── SQL: медиана ₽/м² сопоставимых активных листингов ────────────────────────
#
# percentile_cont(0.5) — тот же идиом, что уже используется в estimator.py для медианных
# ₽/м² трендов (_fetch_price_trend) — устойчив к выбросам В ОТЛИЧИЕ от AVG/min/max: единичный
# аномально дорогой/дешёвый лот не сдвигает медиану заметно.
#
# geo_precision IS DISTINCT FROM 'city' — тот же фильтр, что в estimator.py (#769 Part E):
# исключает листинги с геокодом до центра города (city-centroid fallback без номера дома),
# которые иначе "подмешивались" бы в любой радиус вокруг центра.
#
# price_per_m2 BETWEEN sanity-границы — не бизнес-калибровка, а защита от битых строк
# (см. _PRICE_PER_M2_SANITY_MIN/MAX выше).
#
# bbox-фильтр (lat/lon) — сопоставимые листинги считаются ТОЛЬКО по Екатеринбургу, даже если
# сам продукт уже скрейпит соседние города области (city-sweep): географию location_index
# явно ограничил владелец продукта.
_MEDIAN_PPM2_LOCAL_SQL = text(
    """
    SELECT
        CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS double precision)
            AS median_ppm2,
        count(*) AS n
    FROM listings
    WHERE is_active = true
      AND price_per_m2 IS NOT NULL
      AND price_per_m2 BETWEEN CAST(:price_min AS integer) AND CAST(:price_max AS integer)
      AND (geo_precision IS DISTINCT FROM 'city')
      AND lat BETWEEN CAST(:bbox_south AS double precision)
                  AND CAST(:bbox_north AS double precision)
      AND lon BETWEEN CAST(:bbox_west AS double precision)
                  AND CAST(:bbox_east AS double precision)
      AND ST_DWithin(
          geom::geography,
          ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
          CAST(:radius_m AS double precision)
      )
    """
)

_MEDIAN_PPM2_CITYWIDE_SQL = text(
    """
    SELECT
        CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2) AS double precision)
            AS median_ppm2,
        count(*) AS n
    FROM listings
    WHERE is_active = true
      AND price_per_m2 IS NOT NULL
      AND price_per_m2 BETWEEN CAST(:price_min AS integer) AND CAST(:price_max AS integer)
      AND (geo_precision IS DISTINCT FROM 'city')
      AND lat BETWEEN CAST(:bbox_south AS double precision)
                  AND CAST(:bbox_north AS double precision)
      AND lon BETWEEN CAST(:bbox_west AS double precision)
                  AND CAST(:bbox_east AS double precision)
    """
)

_NEAREST_POI_SQL = text(
    """
    SELECT
        p.name,
        p.category,
        CAST(
            ST_Distance(
                p.geom::geography,
                ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography
            ) AS double precision
        ) AS distance_m
    FROM osm_poi_ekb_local p
    WHERE p.geom IS NOT NULL
      AND ST_DWithin(
          p.geom::geography,
          ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
          CAST(:radius_m AS double precision)
      )
    ORDER BY distance_m ASC
    LIMIT :limit
    """
)


def _local_median_ppm2(db: Any, lat: float, lon: float, radius_m: int) -> tuple[float | None, int]:
    row = (
        db.execute(
            _MEDIAN_PPM2_LOCAL_SQL,
            {
                "lat": lat,
                "lon": lon,
                "radius_m": radius_m,
                "price_min": _PRICE_PER_M2_SANITY_MIN,
                "price_max": _PRICE_PER_M2_SANITY_MAX,
                "bbox_south": _EKB_BBOX_SOUTH,
                "bbox_north": _EKB_BBOX_NORTH,
                "bbox_west": _EKB_BBOX_WEST,
                "bbox_east": _EKB_BBOX_EAST,
            },
        )
        .mappings()
        .first()
    )
    if row is None:
        return None, 0
    median = row["median_ppm2"]
    return (float(median) if median is not None else None), int(row["n"] or 0)


def _citywide_median_ppm2(db: Any) -> tuple[float | None, int]:
    row = (
        db.execute(
            _MEDIAN_PPM2_CITYWIDE_SQL,
            {
                "price_min": _PRICE_PER_M2_SANITY_MIN,
                "price_max": _PRICE_PER_M2_SANITY_MAX,
                "bbox_south": _EKB_BBOX_SOUTH,
                "bbox_north": _EKB_BBOX_NORTH,
                "bbox_west": _EKB_BBOX_WEST,
                "bbox_east": _EKB_BBOX_EAST,
            },
        )
        .mappings()
        .first()
    )
    if row is None:
        return None, 0
    median = row["median_ppm2"]
    return (float(median) if median is not None else None), int(row["n"] or 0)


def _fetch_nearby_poi(
    db: Any, lat: float, lon: float, radius_m: int, top_n: int
) -> tuple[list[NearbyPoi], str]:
    """Top-N POI поблизости — качественная справка «что рядом», не числовой показатель.

    Graceful fallback: osm_poi_ekb_local пуста (рефреш ещё не запускался на этом окружении)
    → ([], "unavailable") вместо 500 или сфабрикованного списка.
    """
    total = db.execute(text("SELECT count(*) FROM osm_poi_ekb_local")).scalar() or 0
    if total == 0:
        logger.warning(
            "location_index: osm_poi_ekb_local is empty (refresh job not yet run on this "
            "environment) — nearby_poi unavailable, no fabricated factors"
        )
        return [], "unavailable"

    rows = (
        db.execute(
            _NEAREST_POI_SQL,
            {"lat": lat, "lon": lon, "radius_m": radius_m, "limit": top_n * 10},
        )
        .mappings()
        .all()
    )

    ranked: list[tuple[float, NearbyPoi]] = []
    for row in rows:
        distance_m = float(row["distance_m"])
        category = row["category"] or "default"
        weight = (1.0 / (distance_m + 100.0)) * _category_weight(category)
        ranked.append(
            (
                weight,
                NearbyPoi(poi_type=category, name=row["name"], distance_m=round(distance_m, 1)),
            )
        )

    ranked.sort(key=lambda pair: pair[0], reverse=True)
    return [poi for _weight, poi in ranked[:top_n]], "ok"


def compute_location_index(
    db: Any,
    lat: float,
    lon: float,
    *,
    radius_m: int | None = None,
    poi_radius_m: int = DEFAULT_POI_RADIUS_M,
    poi_top_n: int = DEFAULT_POI_TOP_N,
) -> LocationIndexResult:
    """Посчитать location index для координат (lat, lon).

    location_index_pct = (медиана ₽/м² сопоставимых активных листингов в радиусе точки −
    медиана ₽/м² по всему ЕКБ) / медиана по ЕКБ * 100. Радиус — лестница RADIUS_LADDER_M
    (расширяется, пока выборка не наберёт MIN_SAMPLE_SIZE), если явный radius_m не передан
    (тогда используется РОВНО он, без расширения — для отладки/тестов).

    Args:
        db: SQLAlchemy Session.
        lat: широта целевой точки.
        lon: долгота целевой точки.
        radius_m: явный радиус в метрах — если задан, лестница не используется.
        poi_radius_m: радиус для качественного списка «что рядом» (независим от индекса).
        poi_top_n: сколько POI показать в «что рядом».

    Returns:
        LocationIndexResult со status:
          - "out_of_coverage" — точка вне bbox Екатеринбурга, ничего не считаем.
          - "insufficient_data" — даже на максимальном радиусе сопоставимых листингов
            меньше MIN_SAMPLE_SIZE (или городская выборка-эталон сама вырождена).
          - "ok" — location_index_pct надёжен.
    """
    if not _in_ekb_bbox(lat, lon):
        logger.info(
            "location_index: lat=%.5f lon=%.5f outside EKB coverage bbox — out_of_coverage",
            lat,
            lon,
        )
        return LocationIndexResult(
            status="out_of_coverage",
            location_index_pct=None,
            local_median_price_per_m2=None,
            city_median_price_per_m2=None,
            sample_size=0,
            radius_m=radius_m or RADIUS_LADDER_M[0],
            nearby_poi=[],
            poi_status="unavailable",
        )

    nearby_poi, poi_status = _fetch_nearby_poi(db, lat, lon, poi_radius_m, poi_top_n)

    city_median, city_n = _citywide_median_ppm2(db)
    if city_median is None or city_n < MIN_SAMPLE_SIZE:
        logger.warning(
            "location_index: citywide reference sample too small (n=%d) — insufficient_data",
            city_n,
        )
        return LocationIndexResult(
            status="insufficient_data",
            location_index_pct=None,
            local_median_price_per_m2=None,
            city_median_price_per_m2=(round(city_median) if city_median is not None else None),
            sample_size=city_n,
            radius_m=radius_m or RADIUS_LADDER_M[-1],
            nearby_poi=nearby_poi,
            poi_status=poi_status,
        )

    radii = [radius_m] if radius_m is not None else list(RADIUS_LADDER_M)
    local_median: float | None = None
    sample_size = 0
    used_radius = radii[-1]
    for r in radii:
        local_median, sample_size = _local_median_ppm2(db, lat, lon, r)
        used_radius = r
        if sample_size >= MIN_SAMPLE_SIZE:
            break

    if local_median is None or sample_size < MIN_SAMPLE_SIZE:
        logger.info(
            "location_index: lat=%.5f lon=%.5f sample=%d < MIN_SAMPLE_SIZE=%d up to "
            "radius=%dm — insufficient_data",
            lat,
            lon,
            sample_size,
            MIN_SAMPLE_SIZE,
            used_radius,
        )
        return LocationIndexResult(
            status="insufficient_data",
            location_index_pct=None,
            local_median_price_per_m2=None,
            city_median_price_per_m2=round(city_median),
            sample_size=sample_size,
            radius_m=used_radius,
            nearby_poi=nearby_poi,
            poi_status=poi_status,
        )

    pct = _pct_deviation(local_median, city_median)
    logger.debug(
        "location_index: lat=%.5f lon=%.5f radius=%dm n=%d local=%d city=%d pct=%.1f",
        lat,
        lon,
        used_radius,
        sample_size,
        round(local_median),
        round(city_median),
        pct,
    )
    return LocationIndexResult(
        status="ok",
        location_index_pct=pct,
        local_median_price_per_m2=round(local_median),
        city_median_price_per_m2=round(city_median),
        sample_size=sample_size,
        radius_m=used_radius,
        nearby_poi=nearby_poi,
        poi_status=poi_status,
    )