"""§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-данные; вызывающий # может добавить). Держим явной константой для детерминизма сетки. _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, )