diff --git a/backend/app/schemas/recommend.py b/backend/app/schemas/recommend.py index 2de03ad2..62e6ba82 100644 --- a/backend/app/schemas/recommend.py +++ b/backend/app/schemas/recommend.py @@ -62,13 +62,36 @@ class RecommendComparable(BaseModel): buildings_n: int | None = None +class RecommendReason(BaseModel): + """§16 структурная причина одного советующего пункта overlay (#983, 953-B). ADVISORY. + + Объясняет, ПОЧЕМУ пункт попал в совет, ЧТО на него повлияло и ЧТО его изменило бы: + • `why` — RU-предложение (сегмент, deficit_index на горизонте, спрос/предложение). + • `drivers` — драйверы сигнала ([{factor, value, direction}]). + • `rejected` — отвергнутые альтернативы из РАНКИНГА (runner-up'ы / негатив-дефицит). + • `what_would_change` — рычаги §9.x, которые сдвинули бы индекс (ставка/предложение/ + горизонт), полными RU-предложениями. + `advisory` ВСЕГДА True (наследует advisory-статус §9.x). Все поля схемы — для + explainability, НЕ основание для инвест-решения. Это «контракт-намёк»: overlay + пробрасывает причины как открытые dict'ы (см. RecommendForecastSegment.reason). + """ + + why: str + drivers: list[dict[str, Any]] = [] + rejected: list[dict[str, Any]] = [] + what_would_change: list[str] = [] + confidence: Literal["high", "medium", "low"] + advisory: bool + + class RecommendForecastSegment(BaseModel): """Одна ранжированная ячейка §9.7 forecast-overlay (#982). ADVISORY. `deficit_index`: в demand_supply-режиме ∈ [−1,+1] (>0 строить / <0 затоварка, supply-based); в demand_only-режиме ∈ (0,1] = ПРОКСИ относительной силы спроса (NOT supply-based — геометрии участка нет). `balance_units` (demand−supply) - None в demand_only (предложение неизмеримо без cad_num). + None в demand_only (предложение неизмеримо без cad_num). `reason` (#983, §16) — + структурная причина ячейки (RecommendReason-shaped dict) или None. """ bucket: str @@ -76,16 +99,25 @@ class RecommendForecastSegment(BaseModel): deficit_index: float balance_units: float | None = None confidence: Literal["high", "medium", "low"] + reason: dict[str, Any] | None = None class RecommendForecastOverlay(BaseModel): - """§9.7 СОВЕТУЮЩИЙ forecast-overlay поверх живого микса (#982, 953-A). + """§9.7 СОВЕТУЮЩИЙ forecast-overlay поверх живого микса (#982, 953-A; #983, 953-B). Кладётся ТОЛЬКО под scope["forecast"] при заданном horizon_months; живой микс (4 поля RecommendMixOutput) НЕ затрагивается. `advisory` ВСЕГДА True (наследует advisory-статус §9.x — не основание для инвест-решения). `mode`: demand_supply (cad_num задан — supply/конкуренты учтены) или demand_only (cad_num=None — только темп спроса; см. warnings). + + #983 (953-B) — ADDITIVE-расширения поверх #982 (старые поля/поведение НЕ тронуты): + • `class_reco` — §10.2 рекомендация класса (сильнейший агрегатный дефицит) + §16 + reason, либо None (нет ранжированных данных). + • `usp` — §10.5 USP-из-дефицитов: top-K самых недообеспеченных сегментов как + «дефицит формата X — стройте его» (пустой при отсутствии данных). + • `commercial` — §10.4 советующий коммерческий сигнал (доля коммерции) или + degraded-honest {available: False, caveat, advisory} при тонких данных, либо None. """ horizon_months: int @@ -93,6 +125,9 @@ class RecommendForecastOverlay(BaseModel): advisory: bool ranked_segments: list[RecommendForecastSegment] warnings: list[str] + class_reco: dict[str, Any] | None = None + usp: list[dict[str, Any]] = [] + commercial: dict[str, Any] | None = None class RecommendMixOutput(BaseModel): diff --git a/backend/app/services/forecasting/recommendation.py b/backend/app/services/forecasting/recommendation.py index bac5f974..1b55056e 100644 --- a/backend/app/services/forecasting/recommendation.py +++ b/backend/app/services/forecasting/recommendation.py @@ -1,10 +1,17 @@ -"""§9.7 forecast-overlay для живого квартирография-рекомендатора (#982, 953-A). +"""§9.7 forecast-overlay для живого квартирография-рекомендатора (#982, 953-A; #983, 953-B). Это ТОНКИЙ СОВЕТУЮЩИЙ мост между ЖИВЫМ `analytics_queries.recommend_mix` и форсайт-движком §9.x (#980/#981). Своей §9.x-математики НЕ пересобирает — собирает ответ из уже-смерженных сервисов и форматирует под schemas.recommend overlay-контракт. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM; своего SQL НЕТ. +#983 (953-B, EPIC 9) ADDITIVE-расширяет overlay поверх #982 (старые поля/поведение +НЕ тронуты): §10.2 рекомендация класса (`class_reco`), §10.5 USP-из-дефицитов (`usp`), +§10.4 советующий коммерческий сигнал (`commercial`, degraded-honest) и §16 структурная +причина (`reason`) на КАЖДОМ ранжированном сегменте + на class_reco/usp. §16-причины +ДЕРИВИРУЮТСЯ из ранкинга what_to_build (отвергнутые альтернативы = runner-up'ы / +негатив-дефицит ячейки) — поэтому маршрут через rank_segments корректен и для §16. + ADVISORY (КРИТИЧНО — честность, зеркало what_to_build.py): overlay наследует advisory-статус §9.x (движок не провалидирован до бэктеста #951). Поэтому `advisory` ВСЕГДА True, confidence ≤ 'medium' в demand_supply-режиме и 'low' в @@ -72,6 +79,34 @@ _CLASS_TO_FORECAST: dict[str, str] = { "Economy": "эконом", } +# ── #983 named-константы (§10/§16) ───────────────────────────────────────────── + +# §10.5 USP-из-дефицитов: сколько самых недообеспеченных сегментов превращаем в +# USP-пункты «дефицит формата X — стройте его». 3 = верхушка сигнала, не «простыня». +_USP_TOP_K: int = 3 + +# §10.4 коммерция: premise_kind, которым пробуем измерить нежилой сток (вокабуляр +# objective_lots — «нежилое»). objective покрывает в основном жильё, поэтому обычно +# выборка тонкая → degraded-honest путь (см. _commercial_signal). НЕ фабрикуем число. +_COMMERCIAL_PREMISE_KIND: str = "нежилое" + +# §10.4: минимум нежилых лотов в выборке, ниже которого сигнал считаем недостаточным +# (degraded-honest). Зеркало духа market_metrics порогов: мало лотов → ненадёжно. +_COMMERCIAL_MIN_LOTS: int = 30 + +# §16 «что изменило бы индекс» — рычаги §9.x полными RU-предложениями (микрокопия: +# п.п., ≥). Templated-константы (НЕ LLM): ставка §9.4/§9.6, скрытый запас §9.3 Layer2, +# горизонт. Шаблон с {horizon} подставляет текущий горизонт вызова. +_LEVER_RATE: str = ( + "Рост ключевой ставки на ≥1 п.п. снижает нормализованный спрос (§9.4) → индекс падает." +) +_LEVER_SUPPLY: str = ( + "Выход скрытого запаса (Layer2, §9.3) в радиусе участка → предложение растёт → индекс падает." +) +_LEVER_HORIZON_TMPL: str = ( + "Сужение горизонта с {horizon} до 6 мес → меньше поглощённого спроса → индекс может вырасти." +) + def map_room_bucket(forecast_bucket: str | None) -> str | None: """Forecast room-bucket → live bucket-id. PURE. Неизвестное/None → None.""" @@ -99,6 +134,271 @@ def map_class(target_class: str | None) -> str | None: return _CLASS_TO_FORECAST.get(target_class) +# ── #983 PURE-билдеры §10/§16 — без БД, полностью юнит-тестируемы ────────────── +# Работают на live segment-dict'ах (shape _demand_*_overlay: bucket / obj_class / +# deficit_index / balance_units / confidence). `all_ranked` — DESC-проекция ранкинга +# what_to_build; из неё §16 деривирует отвергнутые альтернативы (runner-up'ы). + + +def _segment_label(segment: dict[str, Any]) -> str: + """Человекочитаемая метка ячейки для RU-микрокопии §16/§10. PURE. + + « (<класс>)» если класс задан, иначе «». bucket — уже живой + pretty-id (например «3-2-к»); None-bucket → «формат» (не падаем на мусоре). + """ + bucket = segment.get("bucket") or "формат" + obj_class = segment.get("obj_class") + return f"{bucket} ({obj_class})" if obj_class else str(bucket) + + +def _signed(value: float) -> str: + """'+0.62' / '−0.18' — знак для RU-микрокопии индекса (минус — типографский). PURE.""" + return f"+{value:.2f}" if value >= 0 else f"−{abs(value):.2f}" + + +def _direction(value: float) -> str: + """Направление драйвера для §16 ('+' рост сигнала / '−' давление вниз). PURE.""" + return "+" if value >= 0 else "−" + + +def _build_reason( + segment: dict[str, Any], + deficit_index: float, + all_ranked: list[dict[str, Any]], + *, + horizon_months: int, + demand_only: bool = False, + demand: float | None = None, + supply: float | None = None, +) -> dict[str, Any]: + """§16 структурная причина одной ячейки (RecommendReason-shaped dict). PURE. + + Собирает why/drivers/rejected/what_would_change/confidence/advisory: + • why — RU-предложение с deficit_index на горизонте (+ спрос/предложение если + известны; в demand_only оговаривает, что предложение НЕ учтено). + • drivers — deficit_index (всегда) + balance_units (если не None) с direction. + • rejected — ДЕРИВИРУЕТСЯ из `all_ranked`: прочие ячейки (runner-up'ы + + негатив-дефицит) — вот ПОЧЕМУ маршрут через rank_segments корректен для §16. + Негативный дефицит помечаем «затоварка», прочее — «слабее сигнал». + • what_would_change — templated рычаги §9.x (ставка/предложение/горизонт). + `confidence` НАСЛЕДУЕТСЯ из самой ячейки (передаётся в segment). advisory=True. + + Args: + segment: live segment-dict рассматриваемой ячейки (с confidence/bucket/класс). + deficit_index: индекс ячейки (для why/drivers — отдельным аргументом, т.к. + может быть уже округлён вызывающим). + all_ranked: DESC-проекция всего ранкинга (для деривации rejected). + horizon_months: горизонт прогноза (мес) — для why и horizon-рычага. + demand_only: True → why оговаривает, что supply не учтён (demand_only-режим). + demand: оценка спроса (ед.) если известна — иначе опускаем из why. + supply: оценка предложения (ед.) если известна — иначе опускаем из why. + + Returns: + RecommendReason-shaped dict. + """ + label = _segment_label(segment) + why = ( + f"Сегмент {label}: deficit_index {_signed(deficit_index)} на горизонте {horizon_months} мес" + ) + if demand is not None and supply is not None: + why += f" (спрос ~{demand:.0f}, предложение ~{supply:.0f})" + elif demand_only: + why += " (прокси силы спроса; предложение участка НЕ учтено — нет геометрии)" + why += "." + + drivers: list[dict[str, Any]] = [ + { + "factor": "deficit_index", + "value": round(deficit_index, 2), + "direction": _direction(deficit_index), + } + ] + balance_units = segment.get("balance_units") + if balance_units is not None: + drivers.append( + { + "factor": "balance_units", + "value": round(float(balance_units), 1), + "direction": _direction(float(balance_units)), + } + ) + + rejected = _rejected_alternatives(segment, all_ranked) + + what_would_change = [ + _LEVER_RATE, + _LEVER_SUPPLY, + _LEVER_HORIZON_TMPL.format(horizon=horizon_months), + ] + + return { + "why": why, + "drivers": drivers, + "rejected": rejected, + "what_would_change": what_would_change, + "confidence": segment.get("confidence", "low"), + "advisory": True, + } + + +def _rejected_alternatives( + chosen: dict[str, Any], all_ranked: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """§16 отвергнутые альтернативы из РАНКИНГА (всё, кроме выбранной ячейки). PURE. + + Деривируем из `all_ranked` (DESC): каждая ДРУГАЯ ячейка — отвергнутая альтернатива. + Идентичность ячейки — по (bucket, obj_class) (live-ключ overlay). reason-ярлык: + deficit_index < 0 → «затоварка», иначе «слабее сигнал». Сохраняем DESC-порядок + ранкинга (детерминированно). Самоисключение по ключу — НЕ по object identity. + """ + chosen_key = (chosen.get("bucket"), chosen.get("obj_class")) + rejected: list[dict[str, Any]] = [] + for seg in all_ranked: + if (seg.get("bucket"), seg.get("obj_class")) == chosen_key: + continue + di = seg.get("deficit_index") + di_f = float(di) if di is not None else 0.0 + rejected.append( + { + "alternative": _segment_label(seg), + "deficit_index": round(di_f, 2), + "reason": "затоварка" if di_f < 0 else "слабее сигнал", + } + ) + return rejected + + +def _recommend_class( + all_ranked: list[dict[str, Any]], *, horizon_months: int, demand_only: bool = False +) -> dict[str, Any] | None: + """§10.2 рекомендация класса — сильнейший агрегатный дефицит по классам. PURE. + + Агрегируем `deficit_index` per `obj_class` (среднее по его room-bucket'ам), + выбираем класс с сильнейшим средним дефицитом. К рекомендации прикрепляем §16 + `_build_reason` (как для синтетической class-level ячейки; rejected — прочие + классы тем же механизмом). None если нет ранжированных данных или ни у одной + ячейки нет obj_class (нечего агрегировать — НЕ фабрикуем класс). + + Детерминированно: tie-break по obj_class ASC (стабильно при равных средних). + + Args: + all_ranked: DESC-проекция ранкинга (live segment-dict'ы). + horizon_months: горизонт (для §16 reason). + demand_only: проброс в reason (оговорка supply-excluded). + + Returns: + {obj_class, mean_deficit_index, n_segments, reason} либо None. + """ + if not all_ranked: + return None + + by_class: dict[str, list[float]] = {} + for seg in all_ranked: + obj_class = seg.get("obj_class") + di = seg.get("deficit_index") + if obj_class is None or di is None: + continue + by_class.setdefault(obj_class, []).append(float(di)) + if not by_class: + return None + + means = {cls: sum(vals) / len(vals) for cls, vals in by_class.items()} + # Сильнейший средний дефицит; tie-break — obj_class ASC (детерминизм). + best_class = max(means, key=lambda c: (means[c], _neg_str(c))) + best_mean = means[best_class] + + # Синтетическая class-level ячейка для §16: confidence наследуем как лучший + # (max) среди ячеек класса — class-сигнал не слабее своей сильнейшей ячейки. + class_segments = [s for s in all_ranked if s.get("obj_class") == best_class] + confidence = _best_confidence(s.get("confidence") for s in class_segments) + synthetic = {"bucket": "класс", "obj_class": best_class, "confidence": confidence} + # rejected — прочие КЛАССЫ (агрегатные ячейки), не room-bucket'ы. + class_cells = [ + {"bucket": "класс", "obj_class": cls, "deficit_index": m, "confidence": confidence} + for cls, m in means.items() + ] + reason = _build_reason( + synthetic, + best_mean, + class_cells, + horizon_months=horizon_months, + demand_only=demand_only, + ) + + return { + "obj_class": best_class, + "mean_deficit_index": round(best_mean, 3), + "n_segments": len(class_segments), + "reason": reason, + } + + +def _usp_from_deficits( + all_ranked: list[dict[str, Any]], + *, + horizon_months: int, + top_k: int = _USP_TOP_K, + demand_only: bool = False, +) -> list[dict[str, Any]]: + """§10.5 USP-из-дефицитов — top-K самых недообеспеченных сегментов. PURE. + + `all_ranked` уже DESC по deficit_index → первые top_k = сильнейшая недообеспеченность. + Каждый → USP-пункт «Дефицит формата X — стройте его» + §16 reason. Пустой вход → + пустой список (НЕ фабрикуем USP). Берём только ячейки с deficit_index не None. + + Args: + all_ranked: DESC-проекция ранкинга (live segment-dict'ы). + horizon_months: горизонт (для §16 reason). + top_k: сколько верхних сегментов превратить в USP (по умолчанию _USP_TOP_K). + demand_only: проброс в reason (оговорка supply-excluded). + + Returns: + Список USP-пунктов {segment, obj_class, deficit_index, usp_text, reason}. + """ + usp: list[dict[str, Any]] = [] + for seg in all_ranked[:top_k]: + di = seg.get("deficit_index") + if di is None: + continue + usp.append( + { + "segment": seg.get("bucket"), + "obj_class": seg.get("obj_class"), + "deficit_index": round(float(di), 3), + "usp_text": f"Дефицит формата «{_segment_label(seg)}» — стройте его.", + "reason": _build_reason( + seg, + float(di), + all_ranked, + horizon_months=horizon_months, + demand_only=demand_only, + ), + } + ) + return usp + + +def _neg_str(value: str) -> tuple[int, ...]: + """Ключ для ASC-tie-break внутри max() (max берёт «больший» → инвертируем). PURE. + + max(..., key) выбирает максимум; чтобы при равном primary взять лексикографически + МЕНЬШИЙ obj_class, отдаём покодовый кортеж с отрицанием (меньшая строка → больший + ключ). Детерминированно, без локали. + """ + return tuple(-ord(ch) for ch in value) + + +def _best_confidence(values: Any) -> str: + """MAX confidence среди ячеек (class-сигнал не слабее сильнейшей ячейки). PURE. + + Зеркало _CONFIDENCE-vocab (low dict[str, Any]: - """Собрать RecommendForecastOverlay-shaped dict. advisory ВСЕГДА True. PURE.""" + """Собрать RecommendForecastOverlay-shaped dict. advisory ВСЕГДА True. PURE. + + #982-поля (horizon_months/mode/advisory/ranked_segments/warnings) — без изменений. + #983 ADDITIVE-поля (class_reco §10.2 / usp §10.5 / commercial §10.4) — опциональны, + дефолты сохраняют #982-байт-в-байт-форму при пустом расширении (None/[]). + """ return { "horizon_months": horizon_months, "mode": mode, "advisory": True, "ranked_segments": ranked_segments, "warnings": warnings, + "class_reco": class_reco, + "usp": usp if usp is not None else [], + "commercial": commercial, } + + +def _commercial_signal( + db: Session, district: str | None, horizon_months: int +) -> dict[str, Any] | None: + """§10.4 советующий коммерческий сигнал (доля коммерции) — degraded-honest. Graceful. + + Пробует измерить нежилой сток через `compute_market_metrics(premise_kind= + "нежилое")`. objective покрывает в основном жильё → выборка обычно тонкая. Тогда + возвращаем degraded-honest {available: False, caveat, advisory} — НЕ фабрикуем число. + Если данных достаточно (≥ _COMMERCIAL_MIN_LOTS лотов) → советующая оценка доли + коммерции (sell_through_pct как прокси реализованной доли) + §16-подобный reason. + НИКОГДА не бросает: любой сбой движка/импорта → degraded-honest None-сигнал. + + Args: + db: SQLAlchemy sync Session. + district: район (None → ЕКБ-wide). + horizon_months: горизонт (для контекста caveat/reason). + + Returns: + Советующий dict {available, ...} либо None при полном отсутствии движка. + """ + # Локальный импорт — зеркало остального модуля (избегаем import-cycle). + from app.services.site_finder.market_metrics import compute_market_metrics + + caveat = ( + f"коммерция: нет достаточных данных (premise_kind={_COMMERCIAL_PREMISE_KIND}, " + f"горизонт {horizon_months} мес)" + ) + try: + metrics = compute_market_metrics( + db, district=district, premise_kind=_COMMERCIAL_PREMISE_KIND + ) + n_lots = _as_int(getattr(metrics, "n_lots", None)) + sell_through = _as_float(getattr(metrics, "sell_through_pct", None)) + confidence = getattr(metrics, "confidence", "low") + except Exception: + # Движок не поддерживает коммерческий premise_kind / сбой — честный degrade. + logger.exception( + "forecast-overlay commercial: market_metrics failed (district=%s) → degraded", + district, + ) + return {"available": False, "caveat": caveat, "advisory": True} + + # Тонкая выборка (objective ~ жильё) / неизмеримая доля → degraded-honest, без + # фабрикации числа. n_lots None (нечисловой/сбой) тоже трактуем как недостаток. + if n_lots is None or n_lots < _COMMERCIAL_MIN_LOTS or sell_through is None: + logger.info( + "forecast-overlay commercial: thin (district=%s n_lots=%s) → degraded (ADVISORY)", + district, + n_lots, + ) + return {"available": False, "caveat": caveat, "advisory": True} + + # Достаточно данных: советующая оценка реализованной доли коммерции (прокси). + share_pct = round(sell_through, 1) + confidence = confidence if confidence in ("high", "medium", "low") else "low" + return { + "available": True, + "premise_kind": _COMMERCIAL_PREMISE_KIND, + "commercial_share_pct": share_pct, + "n_lots": n_lots, + "confidence": confidence, + "reason": { + "why": ( + f"Коммерция (нежилое): реализованная доля ~{share_pct}% по {n_lots} " + f"лотам на горизонте {horizon_months} мес (прокси спроса на нежилые помещения)." + ), + "drivers": [ + {"factor": "sell_through_pct", "value": share_pct, "direction": "+"}, + {"factor": "n_lots", "value": n_lots, "direction": "+"}, + ], + "rejected": [], + "what_would_change": [_LEVER_SUPPLY], + "confidence": confidence, + "advisory": True, + }, + "advisory": True, + } + + +def _as_int(value: Any) -> int | None: + """Безопасно привести значение к int (None/нечисловое → None). Graceful. PURE. + + Защита §10.4 от мок/мусор-атрибутов (MagicMock < int бросил бы TypeError): любой + нечисловой/bool/сбойный вход → None → degraded-honest путь, без падения. + """ + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return int(value) + + +def _as_float(value: Any) -> float | None: + """Безопасно привести значение к float (None/нечисловое → None). Graceful. PURE.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def _enrich_overlay( + db: Session, + *, + horizon_months: int, + mode: str, + ranked_segments: list[dict[str, Any]], + warnings: list[str], + district: str | None, + demand_only: bool, +) -> dict[str, Any]: + """#983 ADDITIVE-сборка: §16 reason на каждом сегменте + §10.2/§10.5/§10.4. Graceful. + + Мутирует `ranked_segments` IN-PLACE, добавляя `reason` (§16) к каждой ячейке + (rejected деривируется из всего списка). Затем считает class_reco (§10.2), + usp (§10.5) и commercial (§10.4) и собирает overlay через `_overlay`. Расширения + §10.2/§10.5 только при НАЛИЧИИ ранжированных данных (иначе None/[]); commercial + (§10.4) пробуем всегда (он независим от ранкинга, сам degraded-honest). Все + #982-поля сохраняются _overlay'ем без изменений. + + Args: + db: SQLAlchemy sync Session (для §10.4 коммерции). + horizon_months: горизонт прогноза (мес). + mode: "demand_supply" | "demand_only". + ranked_segments: live segment-dict'ы (мутируются — добавляется reason). + warnings: #982-варнинги (пробрасываются как есть). + district: район (для §10.4). + demand_only: True → reason'ы оговаривают, что supply не учтён. + + Returns: + RecommendForecastOverlay-shaped dict с #983-расширениями. + """ + for seg in ranked_segments: + di = seg.get("deficit_index") + if di is None: + continue + seg["reason"] = _build_reason( + seg, + float(di), + ranked_segments, + horizon_months=horizon_months, + demand_only=demand_only, + ) + + class_reco = ( + _recommend_class(ranked_segments, horizon_months=horizon_months, demand_only=demand_only) + if ranked_segments + else None + ) + usp = ( + _usp_from_deficits(ranked_segments, horizon_months=horizon_months, demand_only=demand_only) + if ranked_segments + else [] + ) + commercial = _commercial_signal(db, district, horizon_months) + + return _overlay( + horizon_months, + mode, + ranked_segments, + warnings, + class_reco=class_reco, + usp=usp, + commercial=commercial, + ) diff --git a/backend/tests/services/forecasting/test_recommendation.py b/backend/tests/services/forecasting/test_recommendation.py index c9bad83b..41903812 100644 --- a/backend/tests/services/forecasting/test_recommendation.py +++ b/backend/tests/services/forecasting/test_recommendation.py @@ -1,4 +1,4 @@ -"""Tests для §9.7 forecast-overlay моста (#982, 953-A) recommendation.py. +"""Tests для §9.7 forecast-overlay моста (#982, 953-A; #983, 953-B) recommendation.py. Покрывает: • PURE bridge-таблицы map_room_bucket / map_room_bucket_inverse — обе стороны + @@ -12,6 +12,14 @@ flips, обязательный warning, balance_units None, supply НЕ фабрикуется. - graceful: rank_segments бросает ValueError → пустой ranked_segments + warning, НЕ исключение. + • #983 (953-B) PURE-билдеры §10/§16: + - _build_reason — why/drivers/rejected(из runner-up'ов)/what_would_change shape, + confidence inherited, advisory True. + - _recommend_class — сильнейший агрегатный дефицит; None на пустом. + - _usp_from_deficits — top-K по deficit, RU usp_text. + - _commercial_signal — degraded → available False + caveat, НИКОГДА не crash. + - build_forecast_overlay теперь несёт class_reco/usp/commercial + per-segment reason, + а #982-поведение (ключи/порядок/режимы) НЕ сломано. Mock-based — живой БД не требуют (db = MagicMock; §9.x вызовы патчатся по месту ЛОКАЛЬНОГО импорта внутри функций recommendation.py). @@ -26,6 +34,11 @@ import pytest from app.services.forecasting.recommendation import ( _FORECAST_TO_LIVE_BUCKET, + _USP_TOP_K, + _build_reason, + _commercial_signal, + _recommend_class, + _usp_from_deficits, build_forecast_overlay, map_class, map_room_bucket, @@ -479,3 +492,381 @@ class TestDemandOnlyOverlay: model = RecommendForecastOverlay.model_validate(out) assert model.mode == "demand_only" assert model.advisory is True + + +# ────────────────────────────────────────────────────────────────────────────── +# #983 (953-B) PURE-билдеры §10/§16 +# ────────────────────────────────────────────────────────────────────────────── + + +def _seg( + bucket: str, + deficit_index: float, + *, + obj_class: str | None = "комфорт", + balance_units: float | None = 12.0, + confidence: str = "medium", +) -> dict[str, Any]: + """Live segment-dict (shape _demand_*_overlay) для PURE-билдеров #983.""" + return { + "bucket": bucket, + "obj_class": obj_class, + "deficit_index": deficit_index, + "balance_units": balance_units, + "confidence": confidence, + } + + +# ── §16: _build_reason ───────────────────────────────────────────────────────── + + +class TestBuildReason: + def _ranked(self) -> list[dict[str, Any]]: + # Фиксированный ранкинг-фикстура (DESC): топ + runner-up + негатив-дефицит. + return [ + _seg("1-Студия", 0.62, balance_units=62.0, confidence="medium"), + _seg("3-2-к", 0.20, obj_class="бизнес", confidence="low"), + _seg("4-3-к", -0.18, confidence="medium"), + ] + + def test_why_mentions_segment_index_and_horizon(self) -> None: + ranked = self._ranked() + r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12) + assert "1-Студия" in r["why"] + assert "+0.62" in r["why"] + assert "12 мес" in r["why"] + + def test_drivers_deficit_and_balance_with_direction(self) -> None: + ranked = self._ranked() + r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12) + factors = {d["factor"]: d for d in r["drivers"]} + assert factors["deficit_index"]["value"] == 0.62 + assert factors["deficit_index"]["direction"] == "+" + assert factors["balance_units"]["value"] == 62.0 + assert factors["balance_units"]["direction"] == "+" + + def test_balance_units_omitted_when_none(self) -> None: + # demand_only-ячейка (balance_units None) → драйвера balance_units нет. + seg = _seg("1-Студия", 0.8, balance_units=None, confidence="low") + r = _build_reason(seg, 0.8, [seg], horizon_months=12, demand_only=True) + factors = {d["factor"] for d in r["drivers"]} + assert "balance_units" not in factors + assert "deficit_index" in factors + + def test_rejected_derived_from_runner_ups(self) -> None: + # rejected = ВСЕ прочие ячейки ранкинга (runner-up + негатив-дефицит). + ranked = self._ranked() + r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12) + alts = {x["alternative"] for x in r["rejected"]} + assert "3-2-к (бизнес)" in alts + assert "4-3-к (комфорт)" in alts + # Сама выбранная ячейка не попадает в отвергнутые. + assert "1-Студия (комфорт)" not in alts + + def test_rejected_labels_overstock_vs_weaker(self) -> None: + ranked = self._ranked() + r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12) + by_alt = {x["alternative"]: x["reason"] for x in r["rejected"]} + assert by_alt["3-2-к (бизнес)"] == "слабее сигнал" # положительный, но ниже + assert by_alt["4-3-к (комфорт)"] == "затоварка" # негатив-дефицит + + def test_what_would_change_has_three_levers_with_horizon(self) -> None: + ranked = self._ranked() + r = _build_reason(ranked[0], 0.62, ranked, horizon_months=9) + wwc = r["what_would_change"] + assert len(wwc) == 3 + assert any("ключевой ставки" in s and "п.п." in s for s in wwc) + assert any("Layer2" in s for s in wwc) + assert any("9 до 6 мес" in s for s in wwc) # horizon подставлен + + def test_confidence_inherited_from_segment(self) -> None: + seg = _seg("1-Студия", 0.5, confidence="low") + r = _build_reason(seg, 0.5, [seg], horizon_months=12) + assert r["confidence"] == "low" + + def test_advisory_always_true(self) -> None: + seg = _seg("1-Студия", 0.5) + r = _build_reason(seg, 0.5, [seg], horizon_months=12) + assert r["advisory"] is True + + def test_demand_only_why_notes_supply_excluded(self) -> None: + seg = _seg("1-Студия", 0.9, balance_units=None, confidence="low") + r = _build_reason(seg, 0.9, [seg], horizon_months=12, demand_only=True) + assert "предложение участка НЕ учтено" in r["why"] + + +# ── §10.2: _recommend_class ──────────────────────────────────────────────────── + + +class TestRecommendClass: + def test_picks_strongest_aggregate(self) -> None: + # комфорт: mean(0.6, -0.2)=0.2 ; бизнес: mean(0.5, 0.5)=0.5 → бизнес сильнее. + ranked = [ + _seg("1-Студия", 0.6, obj_class="комфорт"), + _seg("3-2-к", -0.2, obj_class="комфорт"), + _seg("4-3-к", 0.5, obj_class="бизнес"), + _seg("5-80+ м²", 0.5, obj_class="бизнес"), + ] + out = _recommend_class(ranked, horizon_months=12) + assert out is not None + assert out["obj_class"] == "бизнес" + assert out["mean_deficit_index"] == pytest.approx(0.5) + assert out["n_segments"] == 2 + assert out["reason"]["advisory"] is True + + def test_none_on_empty(self) -> None: + assert _recommend_class([], horizon_months=12) is None + + def test_none_when_no_obj_class(self) -> None: + # Ни у одной ячейки нет класса — агрегировать нечего → None (НЕ фабрикуем). + ranked = [_seg("1-Студия", 0.6, obj_class=None)] + assert _recommend_class(ranked, horizon_months=12) is None + + def test_reason_rejected_lists_other_classes(self) -> None: + # комфорт (0.6) > бизнес (0.5) → выбран комфорт, отвергнут бизнес. + ranked = [ + _seg("1-Студия", 0.6, obj_class="комфорт"), + _seg("4-3-к", 0.5, obj_class="бизнес"), + ] + out = _recommend_class(ranked, horizon_months=12) + assert out is not None + assert out["obj_class"] == "комфорт" + alts = {x["alternative"] for x in out["reason"]["rejected"]} + # Отвергнут другой КЛАСС (агрегатная class-level ячейка), не выбранный. + assert any("бизнес" in a for a in alts) + assert not any("комфорт" in a for a in alts) + + +# ── §10.5: _usp_from_deficits ────────────────────────────────────────────────── + + +class TestUspFromDeficits: + def test_top_k_by_deficit(self) -> None: + ranked = [ + _seg("1-Студия", 0.9), + _seg("2-1-к", 0.7), + _seg("3-2-к", 0.5), + _seg("4-3-к", 0.3), + _seg("5-80+ м²", 0.1), + ] + usp = _usp_from_deficits(ranked, horizon_months=12, top_k=3) + assert len(usp) == 3 + # Сохранён DESC-порядок верхушки. + assert [u["segment"] for u in usp] == ["1-Студия", "2-1-к", "3-2-к"] + + def test_default_top_k_is_module_const(self) -> None: + ranked = [_seg(f"b{i}", 1.0 - i * 0.1) for i in range(10)] + usp = _usp_from_deficits(ranked, horizon_months=12) + assert len(usp) == _USP_TOP_K + + def test_usp_text_is_russian_deficit_phrasing(self) -> None: + usp = _usp_from_deficits([_seg("1-Студия", 0.9)], horizon_months=12) + assert usp[0]["usp_text"] == "Дефицит формата «1-Студия (комфорт)» — стройте его." + + def test_each_usp_carries_reason(self) -> None: + usp = _usp_from_deficits([_seg("1-Студия", 0.9)], horizon_months=12) + assert usp[0]["reason"]["advisory"] is True + assert "deficit_index" in {d["factor"] for d in usp[0]["reason"]["drivers"]} + + def test_empty_input_empty_list(self) -> None: + assert _usp_from_deficits([], horizon_months=12) == [] + + +# ── §10.4: _commercial_signal (degraded-honest, never crash) ────────────────── + + +def _mk_commercial_metrics( + n_lots: int, sell_through_pct: float | None, confidence: str = "medium" +) -> MagicMock: + m = MagicMock() + m.n_lots = n_lots + m.sell_through_pct = sell_through_pct + m.confidence = confidence + return m + + +class TestCommercialSignal: + def test_thin_data_degraded_with_caveat(self) -> None: + # Мало нежилых лотов → degraded-honest, БЕЗ фабрикации доли. + with patch(_METRICS, return_value=_mk_commercial_metrics(5, None)): + out = _commercial_signal(MagicMock(), "Ленинский", 12) + assert out is not None + assert out["available"] is False + assert "нет достаточных данных" in out["caveat"] + assert "нежилое" in out["caveat"] + assert out["advisory"] is True + + def test_engine_exception_degraded_never_crash(self) -> None: + # Движок бросает (не поддерживает commercial premise_kind) → честный degrade. + with patch(_METRICS, side_effect=RuntimeError("no premise")): + out = _commercial_signal(MagicMock(), "Ленинский", 12) + assert out is not None + assert out["available"] is False + assert out["advisory"] is True + + def test_magicmock_metrics_does_not_crash(self) -> None: + # Нечисловой n_lots (MagicMock-атрибут) НЕ должен бросать TypeError. + with patch(_METRICS, return_value=MagicMock()): + out = _commercial_signal(MagicMock(), "Ленинский", 12) + assert out is not None + assert out["available"] is False # нечисловой → недостаток данных + + def test_sufficient_data_advisory_share(self) -> None: + # Достаточно лотов + измеримая доля → советующая оценка + reason (НЕ crash). + with patch(_METRICS, return_value=_mk_commercial_metrics(120, 42.5, "medium")): + out = _commercial_signal(MagicMock(), "Ленинский", 12) + assert out is not None + assert out["available"] is True + assert out["commercial_share_pct"] == 42.5 + assert out["n_lots"] == 120 + assert out["confidence"] == "medium" + assert out["reason"]["advisory"] is True + assert out["advisory"] is True + + +# ── build_forecast_overlay: #983 ADDITIVE-расширения присутствуют ───────────── + + +class TestOverlayForecast983Additions: + def _ranking(self) -> WhatToBuildRanking: + return _ranking( + [ + _ranked(room_bucket="Студии 15-30", deficit_index=0.6, obj_class="комфорт"), + _ranked(room_bucket="2-к 45-60", deficit_index=0.2, obj_class="бизнес"), + _ranked(room_bucket="3-к 60-80", deficit_index=-0.3, obj_class="комфорт"), + ] + ) + + def test_overlay_has_new_keys(self) -> None: + with patch(_RANK, return_value=self._ranking()): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + assert "class_reco" in out + assert "usp" in out + assert "commercial" in out + + def test_each_ranked_segment_has_reason(self) -> None: + with patch(_RANK, return_value=self._ranking()): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + assert out["ranked_segments"] + for seg in out["ranked_segments"]: + assert seg["reason"]["advisory"] is True + assert "why" in seg["reason"] + assert "rejected" in seg["reason"] + + def test_class_reco_present_with_ranked_data(self) -> None: + with patch(_RANK, return_value=self._ranking()): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + assert out["class_reco"] is not None + assert out["class_reco"]["obj_class"] in ("комфорт", "бизнес") + + def test_usp_top_k_with_ranked_data(self) -> None: + with patch(_RANK, return_value=self._ranking()): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + assert 1 <= len(out["usp"]) <= _USP_TOP_K + assert out["usp"][0]["segment"] == "1-Студия" # сильнейший дефицит + + def test_commercial_degraded_on_mock_db(self) -> None: + # MagicMock-db → нет реальных нежилых лотов → degraded-honest commercial. + with patch(_RANK, return_value=self._ranking()): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + assert out["commercial"]["available"] is False + assert out["commercial"]["advisory"] is True + + def test_empty_ranking_class_reco_none_usp_empty(self) -> None: + # Пустой ранкинг → class_reco None, usp [], но overlay не падает + commercial есть. + with patch(_RANK, return_value=_ranking([])): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + assert out["class_reco"] is None + assert out["usp"] == [] + assert out["commercial"] is not None + + def test_982_keys_still_present_and_unchanged(self) -> None: + # #982-инвариант: исходные поля overlay на месте + advisory True + mode корректен. + with patch(_RANK, return_value=self._ranking()): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + for key in ("horizon_months", "mode", "advisory", "ranked_segments", "warnings"): + assert key in out + assert out["mode"] == "demand_supply" + assert out["advisory"] is True + assert out["horizon_months"] == 12 + + def test_demand_only_reasons_note_supply_excluded(self) -> None: + # demand_only: per-segment reason оговаривает, что предложение не учтено. + with ( + patch(_METRICS, return_value=_mk_metrics(4.0)), + patch(_GET_MACRO, return_value=[]), + patch(_HOLD, return_value={12: 18.0}), + patch(_NORM, return_value=_mk_coef(1.0)), + patch(_MACRO, return_value=_mk_coef(1.0)), + ): + out = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num=None, + horizon_months=12, + target_class=None, + ) + assert out["ranked_segments"] + assert all( + "предложение участка НЕ учтено" in s["reason"]["why"] + for s in out["ranked_segments"] + ) + + def test_overlay_with_983_fields_validates_against_schema(self) -> None: + from app.schemas.recommend import RecommendForecastOverlay + + with patch(_RANK, return_value=self._ranking()): + out: dict[str, Any] = build_forecast_overlay( + MagicMock(), + district="Ленинский", + cad_num="66:41:0000000:1", + horizon_months=12, + target_class=None, + ) + model = RecommendForecastOverlay.model_validate(out) + assert model.class_reco is not None + assert model.usp + assert model.commercial is not None + assert model.ranked_segments[0].reason is not None