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

"""Location-сервис — district-level индексы первоклассной «локации» (#948 Part B, §8.2).

Промоутит район (district) в queryable сущность `location` (data/sql/146_location.sql),
несущую 4 нормированных индекса:
  • infra        — инфраструктура района
  • competition  — конкуренция (доступное/затоваренное конкурирующее предложение)
  • demand       — спрос (темп продаж)
  • future_supply — будущее давление предложения

ADDITIVE (HARD CONSTRAINT #948B): `district` (string) НЕ рефакторится в FK. Этот
сервис — НОВЫЙ слой РЯДОМ: ключует `location` по `district_name` (joinable по строке),
переиспользуя существующие per-district forecast-функции (они #1129-кэшированы, вызов
per-district дёшев). Существующий код продолжает ходить по district-строкам.

ИСТОЧНИКИ ИНДЕКСОВ (reuse, не изобретаем) + НОРМАЛИЗАЦИЯ к [0,1] (None при отсутствии
данных — НИКОГДА не фабрикуем 0, дух market_metrics.py / future_supply.py):

  • competition ← market_metrics.compute_market_metrics(db, district=…).overstock_index
        overstock_index = долго-экспонируемый непроданный сток / все доступные ед.
        ∈ [0,1] — доля «зависшего» конкурирующего предложения (в продаже > N мес без
        сделки). Это ИСТИННЫЙ сигнал конкуренции = доступное/затоваренное конкурирующее
        предложение: выше = больше конкурирующего стока простаивает = БОЛЬШЕ
        конкурентного давления для нового входящего игрока. УЖЕ нормирован [0,1]
        (доля от доступных) — берём как есть, без пере-нормировки. None когда метрика
        None (нет доступных лотов / пустая выборка).

        ОРТОГОНАЛЬНОСТЬ к demand: competition мерит ДОСТУПНОЕ/непроданное предложение,
        demand — скорость продаж (unit_velocity). Это разные оси (saturation supply vs
        sales velocity), не монотонно связанные. Раньше тут стоял sell_through_pct
        (доля реализованного стока) — но это absorption/heat-сигнал, монотонно
        растущий с продажами (коррелирован с demand=velocity) И инвертированный
        относительно «свободного конкурирующего стока»: высокий sell-through =
        МЕНЬШЕ доступной конкуренции, а не больше. overstock_index — честная,
        ортогональная спросу замена.

  • demand ← market_metrics.compute_market_metrics(db, district=…).unit_velocity
        unit_velocity = ед., проданных в месяц (за окно) ∈ [0, ∞). Это ТОТ ЖЕ сигнал
        спроса, что использует forecasting/demand_supply_forecast (base_pace). Нет
        естественного потолка → насыщающее преобразование (зеркало
        future_supply._saturating_index): min(1.0, velocity / _DEMAND_SATURATION_UPM).
        _DEMAND_SATURATION_UPM = 50 ед./мес — район, продающий ≥50 квартир/мес,
        считаем «максимальный спрос» (index=1.0). None когда unit_velocity None.

  • future_supply ← future_supply.compute_future_supply_pressure(db, district=…).index
        УЖЕ нормирован ∈ [0,1] самим сервисом (насыщение months_of_pressure). Берём
        как есть — НЕ пере-нормируем. None когда index None (нет supply-данных /
        поглощения).

  • infra ← analytics_queries._district_poi_score(db, district_name) /
            analytics_queries._city_avg_poi_score(db)
        _district_poi_score = средний по ЖК района weighted POI-count (метро/медицина
        тяжелее) в радиусе 1000м; _city_avg_poi_score — то же по всему ЕКБ (готовый
        нормализатор, уже существует). Нормализация: district / city, clamp [0,1] —
        район со средней-по-городу инфраструктурой ≈ 0.5..1.0, лучше города → clamp в
        1.0. Это РЕАЛЬНЫЙ per-district infra-сигнал (НЕ NULL-deferred). None когда в
        районе <3 ЖК с POI (_district_poi_score вернул None) ИЛИ нет city-baseline.

DISTRICT SOURCE: 8 официальных АДМИН-районов ЕКБ — через
district_resolver._admin_names(db) (готовый кэшированный запрос
`SELECT district_name FROM ekb_districts WHERE geom IS NOT NULL`; 'не определён' там
нет — и не должно быть). Это канонический district-вокабуляр для наполнения локаций.

CENTROID: derivable из ekb_districts_geom (PostGIS-полигоны, миграция 56) через
ST_Centroid(geom) — наполняется в refresh_locations прямым SQL (район может не иметь
geom-строки → centroid остаётся NULL, graceful).

Conventions: psycopg v3 CAST(:x AS type) — НИКОГДА :x::type; SAVEPOINT per-row в
refresh-цикле (backend.md); logger не print; graceful — сбой одного района не валит
остальные. Детерминированно, без LLM.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.schemas.location import LocationList, LocationOut
from app.services.analytics_queries import _city_avg_poi_score, _district_poi_score
from app.services.site_finder.district_resolver import _admin_names
from app.services.site_finder.future_supply import compute_future_supply_pressure
from app.services.site_finder.market_metrics import compute_market_metrics

logger = logging.getLogger(__name__)

# Насыщение demand-индекса: unit_velocity (ед./мес), при котором index достигает 1.0.
# 50 квартир/мес по району — очень высокий темп для ЕКБ (зеркало дисциплины насыщения
# future_supply._PRESSURE_SATURATION_MONTHS). Tunable. Линейный clamp интерпретируем:
# index=0.5 ⇔ 25 ед./мес.
_DEMAND_SATURATION_UPM: float = 50.0

# Окно метрик рынка (мес) — совпадает с дефолтом compute_market_metrics (§9.2).
_MARKET_WINDOW_MONTHS: int = 6

# Горизонт будущего предложения (мес) — дефолт compute_future_supply_pressure (§9.3).
_FUTURE_SUPPLY_HORIZON_MONTHS: int = 12


@dataclass(frozen=True)
class LocationIndices:
    """4 district-level индекса (#948 §8.2). Каждый ∈ [0,1] или None (нет данных).

    None НИКОГДА не подменяется 0 — «нет данных» и «честный ноль» обязаны различаться
    (дух market_metrics.py graceful-on-thin-data).
    """

    infra_index: float | None
    competition_index: float | None
    demand_index: float | None
    future_supply_index: float | None

    def as_dict(self) -> dict[str, float | None]:
        return {
            "infra_index": _round_or_none(self.infra_index),
            "competition_index": _round_or_none(self.competition_index),
            "demand_index": _round_or_none(self.demand_index),
            "future_supply_index": _round_or_none(self.future_supply_index),
        }


def _round_or_none(value: float | None, digits: int = 4) -> float | None:
    return round(value, digits) if value is not None else None


# ──────────────────────────────────────────────────────────────────────────────
# Pure-нормализация — без БД, полностью юнит-тестируемо.
# Каждая функция graceful: None на входе → None на выходе (не 0).
# ──────────────────────────────────────────────────────────────────────────────


def _clamp01(value: float) -> float:
    """Зажать в [0,1]. PURE."""
    return max(0.0, min(1.0, value))


def normalize_competition(overstock_index: float | None) -> float | None:
    """Индекс конкуренции = доступное/затоваренное конкурирующее предложение → [0,1]. PURE.

    Источник — market_metrics.overstock_index = долго-экспонируемый непроданный сток /
    все доступные ед. ∈ [0,1]: доля «зависшего» конкурирующего предложения. Выше =
    больше конкурирующего стока простаивает = БОЛЬШЕ конкурентного давления для нового
    входящего игрока. Это ИСТИННЫЙ сигнал конкуренции и он ОРТОГОНАЛЕН demand (скорости
    продаж = unit_velocity), а не коррелирован с ним (как был старый sell_through).

    Уже нормирован [0,1] самим market_metrics (доля от доступных) — только clamp как
    защита от артефактов. None → None (нет данных / нет доступных лотов, НЕ 0 — иначе
    «нет конкурирующего стока для замера» неотличимо от «честный ноль затоваривания»).
    """
    if overstock_index is None:
        return None
    return _clamp01(overstock_index)


def normalize_demand(
    unit_velocity: float | None,
    *,
    saturation_upm: float = _DEMAND_SATURATION_UPM,
) -> float | None:
    """Индекс спроса из unit_velocity (ед./мес ∈ [0,∞)) → [0,1]. PURE.

    Насыщающее преобразование (зеркало future_supply._saturating_index): линейный
    clamp min(1.0, velocity / saturation_upm). Монотонно неубывающее, насыщается в 1.0
    на saturation_upm. None → None. velocity=0.0 (валидное измерение «0 продаж/мес») →
    0.0 (а не None — это честный ноль спроса, market_metrics уже отличил его от
    None-«нет выборки»).
    """
    if unit_velocity is None:
        return None
    if saturation_upm <= 0:
        # Деградация без падения: нулевой масштаб → любой положительный спрос = max.
        return 1.0 if unit_velocity > 0 else 0.0
    return _clamp01(unit_velocity / saturation_upm)


def normalize_infra(
    district_poi: float | None,
    city_avg_poi: float | None,
) -> float | None:
    """Инфра-индекс из POI района относительно POI города → [0,1]. PURE.

    district_poi / city_avg_poi, clamp [0,1]: район со средней-по-городу
    инфраструктурой ≈ 1.0 (но clamp удержит лучше-города в 1.0; район беднее города
    → <1.0). None если district_poi None (в районе <3 ЖК с POI) ИЛИ city-baseline
    None/≤0 (нет нормализатора). Никогда не 0-как-заглушка.
    """
    if district_poi is None or city_avg_poi is None or city_avg_poi <= 0:
        return None
    return _clamp01(district_poi / city_avg_poi)


# ──────────────────────────────────────────────────────────────────────────────
# DB-оркестратор индексов — тонкий, graceful. Pure-нормализация выше тестируется без БД.
# ──────────────────────────────────────────────────────────────────────────────


def compute_location_indices(db: Session, district: str) -> LocationIndices:
    """Вычислить 4 district-level индекса для района (#948 §8.2).

    Переиспользует существующие per-district forecast-функции (#1129-кэшированы) и
    нормализует каждый результат к [0,1] (None при отсутствии данных — см. модульный
    docstring про нормализацию каждого индекса). НИКОГДА не фабрикует 0.

    Graceful: если источник бросает — соответствующий индекс = None (logged), остальные
    считаются. Никогда не валит весь расчёт из-за одного сбойного сигнала.

    Args:
        db: SQLAlchemy sync Session.
        district: имя района (одно из 8 админ-районов ЕКБ).

    Returns:
        LocationIndices (всегда; индекс=None там, где данных нет / источник сбоил).
    """
    competition_index: float | None = None
    demand_index: float | None = None
    future_supply_index: float | None = None
    infra_index: float | None = None

    # ── competition + demand ← market_metrics (один вызов, два сигнала) ──────────
    try:
        metrics = compute_market_metrics(
            db, district=district, window_months=_MARKET_WINDOW_MONTHS
        )
        # competition ← overstock_index (доступное/затоваренное конкурирующее
        # предложение) — ортогонален demand. НЕ sell_through (тот коррелирован с
        # velocity и инвертирован относительно «свободной конкуренции»).
        competition_index = normalize_competition(metrics.overstock_index)
        demand_index = normalize_demand(metrics.unit_velocity)
    except Exception:
        logger.exception(
            "location_indices: market_metrics failed (district=%s) → competition/demand None",
            district,
        )

    # ── future_supply ← future_supply (уже 0..1) ────────────────────────────────
    try:
        fsp = compute_future_supply_pressure(
            db, district=district, horizon_months=_FUTURE_SUPPLY_HORIZON_MONTHS
        )
        # index уже нормирован [0,1] самим сервисом — берём как есть (None passthrough).
        future_supply_index = fsp.index
    except Exception:
        logger.exception(
            "location_indices: future_supply failed (district=%s) → future_supply None",
            district,
        )

    # ── infra ← district POI / city POI ─────────────────────────────────────────
    try:
        district_poi = _district_poi_score(db, district_name=district)
        city_poi = _city_avg_poi_score(db)
        infra_index = normalize_infra(district_poi, city_poi)
    except Exception:
        logger.exception(
            "location_indices: infra (POI) failed (district=%s) → infra None", district
        )

    logger.info(
        "location_indices: district=%s infra=%s competition=%s demand=%s future_supply=%s",
        district,
        _round_or_none(infra_index),
        _round_or_none(competition_index),
        _round_or_none(demand_index),
        _round_or_none(future_supply_index),
    )

    return LocationIndices(
        infra_index=infra_index,
        competition_index=competition_index,
        demand_index=demand_index,
        future_supply_index=future_supply_index,
    )


# ──────────────────────────────────────────────────────────────────────────────
# Refresh — upsert индексов по всем районам (SAVEPOINT per-row, CAST, ON CONFLICT).
# ──────────────────────────────────────────────────────────────────────────────

# centroid derive из ekb_districts_geom: ST_Centroid полигона района. Подзапрос —
# район может не иметь geom-строки → centroid NULL (graceful). ON CONFLICT по
# district_name (UNIQUE) обновляет индексы + centroid + indices_computed_at.
# psycopg v3: ВСЕ binds через CAST(:x AS type) — НИКОГДА :x::type.
_UPSERT_LOCATION_SQL = text(
    """
    INSERT INTO location (
        district_name, region, centroid,
        infra_index, competition_index, demand_index, future_supply_index,
        indices_computed_at, updated_at
    ) VALUES (
        CAST(:district_name AS text),
        CAST(:region AS text),
        (
            SELECT ST_Centroid(g.geom)
              FROM ekb_districts_geom g
             WHERE g.district_name = CAST(:district_name AS text)
             LIMIT 1
        ),
        CAST(:infra_index AS numeric),
        CAST(:competition_index AS numeric),
        CAST(:demand_index AS numeric),
        CAST(:future_supply_index AS numeric),
        now(),
        now()
    )
    ON CONFLICT (district_name) DO UPDATE SET
        region              = EXCLUDED.region,
        centroid            = EXCLUDED.centroid,
        infra_index         = EXCLUDED.infra_index,
        competition_index   = EXCLUDED.competition_index,
        demand_index        = EXCLUDED.demand_index,
        future_supply_index = EXCLUDED.future_supply_index,
        indices_computed_at = EXCLUDED.indices_computed_at,
        updated_at          = now()
    """
)


def refresh_locations(db: Session, *, region: str | None = None) -> dict[str, Any]:
    """Пересчитать + upsert-нуть индексы по всем 8 админ-районам ЕКБ (#948 §8.2).

    Итерирует district-источник (district_resolver._admin_names → ekb_districts),
    для каждого считает индексы (compute_location_indices) и апсертит в `location`.
    Идемпотентно: ON CONFLICT (district_name) обновляет индексы + centroid +
    indices_computed_at (повторный прогон даёт тот же результат при тех же данных).

    SAVEPOINT per-row (backend.md): сбойный район откатывается изолированно через
    begin_nested, не роняя всю транзакцию и не сбивая счётчики (#261/pzz_loader bug).
    db.commit() один раз в конце.

    Args:
        db: SQLAlchemy sync Session.
        region: метка региона в колонку `location.region` (None = ЕКБ/Свердл.).

    Returns:
        Счётчики: {"districts": N, "upserted": M, "failed": K}.
    """
    districts = sorted(_admin_names(db))
    if not districts:
        logger.warning(
            "refresh_locations: district source empty (ekb_districts geom?) — nothing to refresh"
        )
        return {"districts": 0, "upserted": 0, "failed": 0}

    upserted = 0
    failed = 0
    for district in districts:
        try:
            indices = compute_location_indices(db, district)
            with db.begin_nested():
                db.execute(
                    _UPSERT_LOCATION_SQL,
                    {
                        "district_name": district,
                        "region": region,
                        "infra_index": indices.infra_index,
                        "competition_index": indices.competition_index,
                        "demand_index": indices.demand_index,
                        "future_supply_index": indices.future_supply_index,
                    },
                )
            upserted += 1
        except Exception as e:
            failed += 1
            logger.warning("refresh_locations: district=%s upsert failed: %s", district, e)

    db.commit()
    logger.info(
        "refresh_locations: districts=%d upserted=%d failed=%d",
        len(districts),
        upserted,
        failed,
    )
    return {"districts": len(districts), "upserted": upserted, "failed": failed}


# ──────────────────────────────────────────────────────────────────────────────
# Read-helpers для GET-API (read-only сущность — без CRUD).
# centroid не отдаём бинарём: ST_X/ST_Y → lon/lat (NULL если centroid NULL).
# ──────────────────────────────────────────────────────────────────────────────

_SELECT_COLS = """
    id, district_name, region,
    ST_X(centroid) AS lon, ST_Y(centroid) AS lat,
    infra_index, competition_index, demand_index, future_supply_index,
    indices_computed_at, created_at, updated_at
"""


def _row_to_out(r: Any) -> LocationOut:
    return LocationOut(
        id=int(r["id"]),
        district_name=r["district_name"],
        region=r["region"],
        lon=float(r["lon"]) if r["lon"] is not None else None,
        lat=float(r["lat"]) if r["lat"] is not None else None,
        infra_index=float(r["infra_index"]) if r["infra_index"] is not None else None,
        competition_index=(
            float(r["competition_index"]) if r["competition_index"] is not None else None
        ),
        demand_index=float(r["demand_index"]) if r["demand_index"] is not None else None,
        future_supply_index=(
            float(r["future_supply_index"]) if r["future_supply_index"] is not None else None
        ),
        indices_computed_at=r["indices_computed_at"],
        created_at=r["created_at"],
        updated_at=r["updated_at"],
    )


def list_locations(db: Any) -> LocationList:
    """Все локации с индексами (8 районов — пагинация не нужна, отдаём целиком)."""
    rows = (
        db.execute(text(f"SELECT {_SELECT_COLS} FROM location ORDER BY district_name"))
        .mappings()
        .all()
    )
    out = [_row_to_out(r) for r in rows]
    return LocationList(total=len(out), rows=out)


def get_location(db: Any, district_name: str) -> LocationOut | None:
    """Вернуть одну локацию по district_name. None если нет (роутер → 404)."""
    row = (
        db.execute(
            text(f"SELECT {_SELECT_COLS} FROM location WHERE district_name = :dn"),
            {"dn": district_name},
        )
        .mappings()
        .first()
    )
    if row is None:
        return None
    return _row_to_out(row)