gendesign/backend/app/services/forecasting/what_to_build.py

"""§9.7 «что строить» — ранкер сегментной сетки по индексу дефицита (ТЗ §9.7).

#981 (952-B, Site Finder v2 / «GG-форсайт» ТЗ §9.7), EPIC 8 «что строить +
доступность». Это **СБОРОЧНЫЙ слой над #980**: прогоняет уже-смерженный
центральный движок `compute_demand_supply_forecast` (§9.8) по СЕТКЕ сегментов
(класс × room_bucket [× price_bucket]) для участка/горизонта и РАНЖИРУЕТ ячейки по
`deficit_index` (по убыванию = сила сигнала «строить»). Это продуктовый ответ §9.7
«строить студии? двушки? какой класс?». Своей §9.x-математики НЕ пересобирает —
вызывает #980 per-cell. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM; своего SQL НЕТ.

ИНТЕРПРЕТАЦИЯ ИНДЕКСА (наследуется от #980 _deficit_index, см. его docstring):
  • deficit_index > 0  → НЕДОобеспечен (спроса больше предложения) = СИГНАЛ СТРОИТЬ.
  • deficit_index = 0  → баланс спрос↔предложение.
  • deficit_index < 0  → ЗАТОВАРКА (предложения больше спроса) = ИЗБЕГАТЬ.
  • deficit_index None → тонкие данные (предложение неизмеримо) — ячейка ОТБРАСЫВАЕТСЯ.
Ранжируем по убыванию: топ списка = самый сильный сигнал «строить именно это».

ADVISORY (КРИТИЧНО — честность): КАЖДАЯ ячейка наследует advisory-статус #980 и его
жёсткий потолок confidence ≤ 'medium' (движок не провалидирован до бэктеста #951).
Поэтому и весь ранкинг СОВЕТУЮЩИЙ: `generated_advisory` ВСЕГДА True. Цифры — для
explainability/прототипа, не основание для инвест-решения.

СЕТКА (сознательно НЕ взрываем до 1000 ячеек): основные оси — room_bucket × class
(дефолтная сетка из bucket-вокабуляра sales_series). price_bucket опционален
(по умолчанию один «без фильтра цены» проход — иначе ×4 ячеек ради оси, которую
#980 §9.5 учитывает грубо). Все оси — named-константы; вызывающий может сузить.

room_bucket source-семантика (см. sales_series.SegmentSpec): дефолтные ключи —
Source-B room_area-вокабуляр ('Студии 15-30' / '1-к 30-45' / …), т.к. это вывод
room_area_bucket_of. #980 пробрасывает room_bucket в §9.4/§9.5 profile; для агрегатов
класс/район Source-A ('студия'/'1'/'2'/…) — вызывающий может передать свой набор.

Graceful (дух demand_supply_forecast): пустая сетка / все ячейки None → ПУСТОЙ
ранкинг (не crash). Сбой #980 на отдельной ячейке → ячейка пропускается, остальные
ранжируются. Детерминированно (стабильная сортировка с tie-break по segment-ключу).
"""

from __future__ import annotations

import logging
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any, Literal

from sqlalchemy.orm import Session

from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast
from app.services.forecasting.sales_series import (
    ROOM_AREA_BUCKET_1K,
    ROOM_AREA_BUCKET_2K,
    ROOM_AREA_BUCKET_3K,
    ROOM_AREA_BUCKET_LARGE,
    ROOM_AREA_BUCKET_STUDIO,
    SegmentSpec,
)

logger = logging.getLogger(__name__)

Confidence = Literal["high", "medium", "low"]

# ── Named-константы (дефолтная сетка) ─────────────────────────────────────────

# Дефолтные классы недвижимости (ось «класс»). ЕКБ-вокабуляр Объектива в
# «человеческом» регистре — SQL внутри §9.x нормализует регистр. 3 массовых класса
# (премиум опускаем из дефолта — тонкий сегмент, обычно low-данные; вызывающий
# может добавить). Держим явной константой для детерминизма сетки.
# #1959: нижний массовый класс objective_lots.class — «стандарт», НЕ «эконом»
# (verified на prod: стандарт=483k лотов, эконома НЕТ → раньше эта ячейка матчила
# 0 строк → молча выпадала из сетки; посегментный дефицит её не видел).
_DEFAULT_CLASSES: tuple[str, ...] = ("стандарт", "комфорт", "бизнес")

# Дефолтные room-bucket'ы (ось «формат») — Source-B room_area-вокабуляр
# (вывод room_area_bucket_of). 5 форматов: студии / 1-к / 2-к / 3-к / 80+. Это
# продуктовая ось §9.7 «какой формат строить».
_DEFAULT_ROOM_BUCKETS: tuple[str, ...] = (
    ROOM_AREA_BUCKET_STUDIO,
    ROOM_AREA_BUCKET_1K,
    ROOM_AREA_BUCKET_2K,
    ROOM_AREA_BUCKET_3K,
    ROOM_AREA_BUCKET_LARGE,
)

# Дефолтная ось цены: один проход БЕЗ price-фильтра (None). Сознательно не дробим
# по price_bucket в дефолте — иначе сетка ×4 ради оси, которую #980 §9.5 учитывает
# лишь грубо (price_tier в profile). Вызывающий может передать реальные bucket'ы.
_DEFAULT_PRICE_BUCKETS: tuple[str | None, ...] = (None,)

# Ранг уверенности для tie-break (выше = лучше). Зеркало demand_supply_forecast.
_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}


@dataclass(frozen=True)
class RankedSegment:
    """Одна ячейка сетки §9.7 с её сигналом дефицита (наследует #980 на горизонте).

    `deficit_index` ∈ [−1,+1]: >0 = строить (недообеспечен), <0 = избегать
    (затоварка). Ячейки с deficit_index None (тонкие данные) в ранкинг НЕ попадают
    (отброшены ранкером). `confidence` ≤ 'medium' (наследуется от #980 advisory-cap).
    """

    segment: dict[str, str | None]
    deficit_index: float  # не None в ранкинге (None-ячейки отброшены)
    balance_units: float | None  # demand − supply (>0 дефицит / <0 затоварка)
    confidence: Confidence  # ≤ 'medium' (#980 advisory-cap)
    projected_demand_units: float | None = None  # #1745: спрос на горизонт (квартир)

    def as_dict(self) -> dict[str, Any]:
        return {
            "segment": dict(self.segment),
            "deficit_index": _round_or_none(self.deficit_index, 3),
            "balance_units": _round_or_none(self.balance_units, 1),
            "projected_demand_units": _round_or_none(self.projected_demand_units, 1),
            "confidence": self.confidence,
        }


@dataclass(frozen=True)
class WhatToBuildRanking:
    """§9.7 ранжированный ответ «что строить» для участка/горизонта (ТЗ §9.7).

    `ranked` отсортирован по `deficit_index` УБЫВАНИЕ (топ = сильнейший сигнал
    «строить»); ячейки с deficit_index None отброшены (тонкие данные). Пустой
    `ranked` = вся сетка тонкая / нет данных (graceful, не crash).
    `generated_advisory` ВСЕГДА True — каждая ячейка наследует advisory-статус #980;
    ранкинг СОВЕТУЮЩИЙ, не основание для инвест-решения.
    """

    district: str | None
    cad_num: str
    horizon_months: int
    ranked: list[RankedSegment]  # DESC по deficit_index; None-ячейки отброшены
    n_cells_scanned: int  # сколько ячеек сетки прогнали (включая отброшенные)
    n_cells_ranked: int  # сколько попало в ранкинг (deficit_index не None)
    generated_advisory: bool  # ВСЕГДА True (наследует advisory #980)

    def as_dict(self) -> dict[str, Any]:
        return {
            "district": self.district,
            "cad_num": self.cad_num,
            "horizon_months": self.horizon_months,
            "ranked": [r.as_dict() for r in self.ranked],
            "n_cells_scanned": self.n_cells_scanned,
            "n_cells_ranked": self.n_cells_ranked,
            "generated_advisory": self.generated_advisory,
        }


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


# ──────────────────────────────────────────────────────────────────────────────
# Pure-логика — без БД, полностью юнит-тестируемо.
# ──────────────────────────────────────────────────────────────────────────────


def _rank_key(cell: RankedSegment) -> tuple[float, int, str]:
    """Ключ сортировки ячейки сетки для ранжирования §9.7. PURE.

    Сортируем по (deficit_index DESC, confidence DESC, segment-ключ ASC):
      1. deficit_index — primary (НЕГАТИВ для DESC через sorted без reverse-флага,
         чтобы tie-break'и шли в естественном ASC-порядке): больше дефицит = выше.
      2. confidence — tie-break: при равном индексе выше та ячейка, чьему сигналу
         больше доверия (negate ранга для DESC).
      3. segment-ключ — финальный СТАБИЛЬНЫЙ tie-break (детерминизм при равных
         index+confidence): лексикографический по компактной строке сегмента.

    Используется как `sorted(cells, key=_rank_key)` — БЕЗ reverse: негативы
    primary/secondary дают убывание, а строковый ключ — возрастание. PURE.

    Args:
        cell: ячейка сетки (deficit_index не None — None-ячейки отброшены до сорта).

    Returns:
        Кортеж-ключ (−deficit_index, −confidence_rank, segment_key).
    """
    confidence_rank = _CONFIDENCE_RANK.get(cell.confidence, 0)
    return (-cell.deficit_index, -confidence_rank, _segment_key(cell.segment))


def _segment_key(segment: dict[str, str | None]) -> str:
    """Стабильная компактная строка сегмента для детерминированного tie-break. PURE.

    Конкатенация осей в фиксированном порядке (class|room|district|price); None →
    пустая строка. Не для отображения — только для устойчивой сортировки.
    """
    return "|".join(
        str(segment.get(axis) or "")
        for axis in ("obj_class", "room_bucket", "district", "price_bucket")
    )


def _build_grid(
    *,
    district: str | None,
    classes: Sequence[str],
    room_buckets: Sequence[str],
    price_buckets: Sequence[str | None],
) -> list[SegmentSpec]:
    """Декартова сетка SegmentSpec из осей класс × room × price. PURE.

    По одному SegmentSpec на комбинацию (class, room_bucket, price_bucket); district
    проставляется во все (общий для участка). Порядок детерминирован (class-внешний,
    room-средний, price-внутренний) — стабильный n_cells_scanned. PURE.

    Args:
        district: район участка (общий для всех ячеек).
        classes: ось классов.
        room_buckets: ось форматов (room_bucket).
        price_buckets: ось цены (может содержать None = «без фильтра цены»).

    Returns:
        Список SegmentSpec — len = |classes|×|room_buckets|×|price_buckets|.
    """
    grid: list[SegmentSpec] = []
    for obj_class in classes:
        for room_bucket in room_buckets:
            for price_bucket in price_buckets:
                grid.append(
                    SegmentSpec(
                        obj_class=obj_class,
                        room_bucket=room_bucket,
                        district=district,
                        price_bucket=price_bucket,
                    )
                )
    return grid


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


def rank_segments(
    db: Session,
    *,
    district: str | None,
    cad_num: str,
    horizon_months: int,
    classes: Sequence[str] = _DEFAULT_CLASSES,
    room_buckets: Sequence[str] = _DEFAULT_ROOM_BUCKETS,
    price_buckets: Sequence[str | None] = _DEFAULT_PRICE_BUCKETS,
    rate_path: dict[int, float] | None = None,
) -> WhatToBuildRanking:
    """§9.7 ранжировать сегменты по индексу дефицита для участка/горизонта.

    ADVISORY (каждая ячейка наследует advisory-статус #980, см. module docstring) —
    НЕ подключать в production-эндпоинт. СБОРКА над #980, НЕ пересчёт §9.x.

    Прогоняет `compute_demand_supply_forecast` (#980) по сетке класс × room × price
    (дефолт — _build_grid из named-констант). Для КАЖДОЙ ячейки берёт прогноз на
    `horizon_months`, собирает (segment, deficit_index, balance_units, confidence),
    ОТБРАСЫВАЕТ ячейки с deficit_index None (тонкие данные — предложение неизмеримо),
    сортирует по `deficit_index` УБЫВАНИЕ (tie-break: confidence ↓, затем стабильный
    segment-ключ). Топ списка = сильнейший сигнал «строить именно это».

    Graceful: пустая сетка / все ячейки None → ПУСТОЙ ранкинг (не crash). Сбой #980
    на ячейке → ячейка пропускается, остальные ранжируются. Детерминированно.

    Args:
        db: SQLAlchemy sync Session.
        district: район участка (None → ЕКБ-wide).
        cad_num: кадастровый номер участка — вход #980 (§9.7 конкуренты).
        horizon_months: горизонт прогноза (мес) — ОДИН (продуктовый «на горизонте»).
        classes: ось классов (по умолчанию _DEFAULT_CLASSES).
        room_buckets: ось форматов (по умолчанию _DEFAULT_ROOM_BUCKETS, Source-B).
        price_buckets: ось цены (по умолчанию _DEFAULT_PRICE_BUCKETS = [None]).
        rate_path: сценарный {horizon: key_rate} для #980; None → hold_last_rate.

    Returns:
        WhatToBuildRanking — DESC-ранкинг (всегда; пустой при тонкой сетке).
    """
    grid = _build_grid(
        district=district,
        classes=classes,
        room_buckets=room_buckets,
        price_buckets=price_buckets,
    )

    cells: list[RankedSegment] = []
    for spec in grid:
        cell = _forecast_cell(
            db,
            spec=spec,
            district=district,
            cad_num=cad_num,
            horizon_months=horizon_months,
            rate_path=rate_path,
        )
        if cell is not None:
            cells.append(cell)

    ranked = sorted(cells, key=_rank_key)

    logger.info(
        "what_to_build: district=%s cad_num=%s horizon=%d scanned=%d ranked=%d "
        "top_segment=%s top_deficit=%s (ADVISORY)",
        district,
        cad_num,
        horizon_months,
        len(grid),
        len(ranked),
        ranked[0].segment if ranked else None,
        _round_or_none(ranked[0].deficit_index, 3) if ranked else None,
    )

    return WhatToBuildRanking(
        district=district,
        cad_num=cad_num,
        horizon_months=horizon_months,
        ranked=ranked,
        n_cells_scanned=len(grid),
        n_cells_ranked=len(ranked),
        generated_advisory=True,
    )


def _forecast_cell(
    db: Session,
    *,
    spec: SegmentSpec,
    district: str | None,
    cad_num: str,
    horizon_months: int,
    rate_path: dict[int, float] | None,
) -> RankedSegment | None:
    """Прогнать #980 для ОДНОЙ ячейки сетки → RankedSegment | None. Graceful.

    Вызывает compute_demand_supply_forecast на единственном горизонте, берёт его
    deficit_index/balance_units/confidence. None при: deficit_index None (тонкие
    данные — ячейка вне ранкинга), пустом результате, ИЛИ сбое #980 (не валим всю
    сетку — ячейка просто выпадает). Детерминированно.

    Returns:
        RankedSegment или None (отбрасывается выше по стеку).
    """
    try:
        forecasts = compute_demand_supply_forecast(
            db,
            spec=spec,
            district=district,
            cad_num=cad_num,
            horizons=[horizon_months],
            rate_path=rate_path,
        )
    except Exception:
        logger.exception(
            "what_to_build: forecast failed for cell segment=%s horizon=%d",
            spec.as_dict(),
            horizon_months,
        )
        return None

    if not forecasts:
        return None
    forecast = forecasts[0]
    if forecast.deficit_index is None:
        # Тонкие данные (предложение неизмеримо) — ячейка вне ранкинга (НЕ 0-сигнал).
        return None

    return RankedSegment(
        segment=forecast.segment,
        deficit_index=forecast.deficit_index,
        balance_units=forecast.balance_units,
        confidence=forecast.confidence,
        projected_demand_units=forecast.projected_demand_units,
    )