872 lines
37 KiB
Python
872 lines
37 KiB
Python
"""Tests для §9.7 forecast-overlay моста (#982, 953-A; #983, 953-B) recommendation.py.
|
||
|
||
Покрывает:
|
||
• PURE bridge-таблицы map_room_bucket / map_room_bucket_inverse — обе стороны +
|
||
unknown→None + DRIFT-GUARD (сверка _FORECAST_TO_LIVE_BUCKET с инверсом
|
||
analytics_queries._BUCKET_PRETTY — таблицы НЕ должны разъехаться).
|
||
• map_class — сворачивание классов + None/unknown→None.
|
||
• build_forecast_overlay через @patch:
|
||
- demand_supply (cad_num задан): rank_segments замокан → маппинг RankedSegment →
|
||
segment-dict, advisory True, DESC-порядок сохранён, unknown room-bucket отброшен.
|
||
- demand_only (cad_num=None): market_metrics/demand_norm/macro замоканы → mode
|
||
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).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
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,
|
||
map_room_bucket_inverse,
|
||
)
|
||
from app.services.forecasting.what_to_build import RankedSegment, WhatToBuildRanking
|
||
|
||
# Точки ЛОКАЛЬНОГО импорта внутри функций recommendation.py — патчим источник,
|
||
# чтобы перехватить вызов независимо от того, где импорт исполнится.
|
||
_RANK = "app.services.forecasting.what_to_build.rank_segments"
|
||
_METRICS = "app.services.site_finder.market_metrics.compute_market_metrics"
|
||
_NORM = "app.services.forecasting.demand_normalization.compute_demand_normalization"
|
||
_MACRO = "app.services.forecasting.macro_coefficient.compute_macro_coefficient"
|
||
_GET_MACRO = "app.services.forecasting.macro_series.get_monthly_macro"
|
||
_HOLD = "app.services.forecasting.demand_supply_forecast.hold_last_rate"
|
||
|
||
|
||
# ── PURE: map_room_bucket / map_room_bucket_inverse ───────────────────────────
|
||
|
||
|
||
class TestMapRoomBucket:
|
||
def test_forward_all_five(self) -> None:
|
||
assert map_room_bucket("Студии 15-30") == "1-Студия"
|
||
assert map_room_bucket("1-к 30-45") == "2-1-к"
|
||
assert map_room_bucket("2-к 45-60") == "3-2-к"
|
||
assert map_room_bucket("3-к 60-80") == "4-3-к"
|
||
assert map_room_bucket("80+ м²") == "5-80+ м²"
|
||
|
||
def test_inverse_all_five(self) -> None:
|
||
assert map_room_bucket_inverse("1-Студия") == "Студии 15-30"
|
||
assert map_room_bucket_inverse("2-1-к") == "1-к 30-45"
|
||
assert map_room_bucket_inverse("3-2-к") == "2-к 45-60"
|
||
assert map_room_bucket_inverse("4-3-к") == "3-к 60-80"
|
||
assert map_room_bucket_inverse("5-80+ м²") == "80+ м²"
|
||
|
||
def test_round_trip_forward_then_inverse(self) -> None:
|
||
for forecast in _FORECAST_TO_LIVE_BUCKET:
|
||
assert map_room_bucket_inverse(map_room_bucket(forecast)) == forecast
|
||
|
||
def test_unknown_forward_none(self) -> None:
|
||
assert map_room_bucket("чердак") is None
|
||
|
||
def test_unknown_inverse_none(self) -> None:
|
||
assert map_room_bucket_inverse("9-чердак") is None
|
||
|
||
def test_none_passthrough(self) -> None:
|
||
assert map_room_bucket(None) is None
|
||
assert map_room_bucket_inverse(None) is None
|
||
|
||
|
||
class TestBucketTableDriftGuard:
|
||
"""_FORECAST_TO_LIVE_BUCKET ДОЛЖНА быть точным инверсом analytics_queries._BUCKET_PRETTY.
|
||
|
||
Если кто-то поменяет один словарь и забудет другой (дублируем 5 литералов ради
|
||
разрыва import-цикла) — этот тест падает первым.
|
||
"""
|
||
|
||
def test_is_exact_inverse_of_bucket_pretty(self) -> None:
|
||
from app.services.analytics_queries import _BUCKET_PRETTY
|
||
|
||
expected_inverse = {pretty: bid for bid, pretty in _BUCKET_PRETTY.items()}
|
||
assert _FORECAST_TO_LIVE_BUCKET == expected_inverse
|
||
|
||
def test_same_cardinality(self) -> None:
|
||
from app.services.analytics_queries import _BUCKET_PRETTY
|
||
|
||
assert len(_FORECAST_TO_LIVE_BUCKET) == len(_BUCKET_PRETTY) == 5
|
||
|
||
|
||
# ── PURE: map_class ───────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestMapClass:
|
||
@pytest.mark.parametrize(
|
||
("live", "expected"),
|
||
[
|
||
("Comfort", "комфорт"),
|
||
("Comfort+", "комфорт"),
|
||
("Business", "бизнес"),
|
||
("Elite", "бизнес"),
|
||
("Economy", "эконом"),
|
||
],
|
||
)
|
||
def test_folding(self, live: str, expected: str) -> None:
|
||
assert map_class(live) == expected
|
||
|
||
def test_none_to_none(self) -> None:
|
||
assert map_class(None) is None
|
||
|
||
def test_unknown_to_none(self) -> None:
|
||
assert map_class("Luxury++") is None
|
||
|
||
|
||
# ── Helpers для мок-сегментов demand_supply ───────────────────────────────────
|
||
|
||
|
||
def _ranked(
|
||
*,
|
||
room_bucket: str | None,
|
||
deficit_index: float,
|
||
obj_class: str = "комфорт",
|
||
balance_units: float | None = 12.0,
|
||
confidence: str = "medium",
|
||
) -> RankedSegment:
|
||
return RankedSegment(
|
||
segment={
|
||
"obj_class": obj_class,
|
||
"room_bucket": room_bucket,
|
||
"district": "Ленинский",
|
||
"price_bucket": None,
|
||
},
|
||
deficit_index=deficit_index,
|
||
balance_units=balance_units,
|
||
confidence=confidence, # type: ignore[arg-type]
|
||
)
|
||
|
||
|
||
def _ranking(ranked: list[RankedSegment]) -> WhatToBuildRanking:
|
||
return WhatToBuildRanking(
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
ranked=ranked,
|
||
n_cells_scanned=len(ranked),
|
||
n_cells_ranked=len(ranked),
|
||
generated_advisory=True,
|
||
)
|
||
|
||
|
||
# ── build_forecast_overlay: demand_supply (cad_num задан) ──────────────────────
|
||
|
||
|
||
class TestDemandSupplyOverlay:
|
||
def test_mode_and_advisory(self) -> None:
|
||
ranking = _ranking([_ranked(room_bucket="2-к 45-60", deficit_index=0.5)])
|
||
with patch(_RANK, return_value=ranking):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class="Comfort",
|
||
)
|
||
assert out["mode"] == "demand_supply"
|
||
assert out["advisory"] is True
|
||
assert out["horizon_months"] == 12
|
||
|
||
def test_maps_segment_buckets_to_live(self) -> None:
|
||
ranking = _ranking(
|
||
[
|
||
_ranked(room_bucket="Студии 15-30", deficit_index=0.9),
|
||
_ranked(room_bucket="2-к 45-60", deficit_index=0.3),
|
||
]
|
||
)
|
||
with patch(_RANK, return_value=ranking):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
buckets = [s["bucket"] for s in out["ranked_segments"]]
|
||
assert buckets == ["1-Студия", "3-2-к"]
|
||
|
||
def test_desc_order_preserved(self) -> None:
|
||
# rank_segments уже отдаёт DESC; overlay не переупорядочивает.
|
||
ranking = _ranking(
|
||
[
|
||
_ranked(room_bucket="80+ м²", deficit_index=0.8),
|
||
_ranked(room_bucket="1-к 30-45", deficit_index=0.2),
|
||
_ranked(room_bucket="3-к 60-80", deficit_index=-0.4),
|
||
]
|
||
)
|
||
with patch(_RANK, return_value=ranking):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
indices = [s["deficit_index"] for s in out["ranked_segments"]]
|
||
assert indices == [0.8, 0.2, -0.4]
|
||
|
||
def test_fields_passed_through(self) -> None:
|
||
ranking = _ranking(
|
||
[
|
||
_ranked(
|
||
room_bucket="2-к 45-60",
|
||
deficit_index=0.55,
|
||
obj_class="бизнес",
|
||
balance_units=33.0,
|
||
confidence="medium",
|
||
)
|
||
]
|
||
)
|
||
with patch(_RANK, return_value=ranking):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=9,
|
||
target_class="Business",
|
||
)
|
||
seg = out["ranked_segments"][0]
|
||
assert seg["obj_class"] == "бизнес"
|
||
assert seg["deficit_index"] == 0.55
|
||
assert seg["balance_units"] == 33.0
|
||
assert seg["confidence"] == "medium"
|
||
|
||
def test_unknown_room_bucket_dropped(self) -> None:
|
||
ranking = _ranking(
|
||
[
|
||
_ranked(room_bucket="2-к 45-60", deficit_index=0.5),
|
||
_ranked(room_bucket="мансарда", deficit_index=0.9), # неизвестный → drop
|
||
]
|
||
)
|
||
with patch(_RANK, return_value=ranking):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
buckets = [s["bucket"] for s in out["ranked_segments"]]
|
||
assert buckets == ["3-2-к"]
|
||
|
||
def test_class_mapped_into_rank_kwargs(self) -> None:
|
||
ranking = _ranking([_ranked(room_bucket="2-к 45-60", deficit_index=0.5)])
|
||
with patch(_RANK, return_value=ranking) as mock_rank:
|
||
build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class="Elite",
|
||
)
|
||
# Elite → бизнес, передаётся ранкеру как classes=["бизнес"].
|
||
assert mock_rank.call_args.kwargs["classes"] == ["бизнес"]
|
||
|
||
def test_no_class_omits_classes_kwarg(self) -> None:
|
||
# target_class=None → classes НЕ передаётся (движковый дефолт _DEFAULT_CLASSES).
|
||
ranking = _ranking([_ranked(room_bucket="2-к 45-60", deficit_index=0.5)])
|
||
with patch(_RANK, return_value=ranking) as mock_rank:
|
||
build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
assert "classes" not in mock_rank.call_args.kwargs
|
||
|
||
def test_empty_ranking_yields_warning(self) -> None:
|
||
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["ranked_segments"] == []
|
||
assert out["warnings"]
|
||
|
||
|
||
# ── build_forecast_overlay: graceful (rank_segments бросает) ───────────────────
|
||
|
||
|
||
class TestDemandSupplyGraceful:
|
||
def test_value_error_yields_empty_plus_warning_no_raise(self) -> None:
|
||
with patch(_RANK, side_effect=ValueError("нет геометрии участка")):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num="66:41:0000000:1",
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
assert out["mode"] == "demand_supply"
|
||
assert out["advisory"] is True
|
||
assert out["ranked_segments"] == []
|
||
assert out["warnings"]
|
||
|
||
|
||
# ── build_forecast_overlay: demand_only (cad_num=None) ─────────────────────────
|
||
|
||
|
||
def _mk_metrics(unit_velocity: float | None) -> MagicMock:
|
||
m = MagicMock()
|
||
m.unit_velocity = unit_velocity
|
||
return m
|
||
|
||
|
||
def _mk_coef(coefficient: float) -> MagicMock:
|
||
c = MagicMock()
|
||
c.coefficient = coefficient
|
||
return c
|
||
|
||
|
||
class TestDemandOnlyOverlay:
|
||
def test_mode_flips_when_no_cad_num(self) -> None:
|
||
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["mode"] == "demand_only"
|
||
assert out["advisory"] is True
|
||
|
||
def test_mandatory_supply_warning_present(self) -> None:
|
||
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 any("supply/конкуренты НЕ учтены" in w for w in out["warnings"])
|
||
|
||
def test_balance_units_always_none_no_fabricated_supply(self) -> None:
|
||
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"], "ожидали 5 ранжированных форматов"
|
||
assert all(s["balance_units"] is None for s in out["ranked_segments"])
|
||
|
||
def test_confidence_low_for_all_segments(self) -> None:
|
||
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 all(s["confidence"] == "low" for s in out["ranked_segments"])
|
||
|
||
def test_deficit_index_is_pace_proxy_zero_to_one(self) -> None:
|
||
# Разные §9.5-коэффициенты per вызов → разные pace → deficit_index = pace/max ∈ (0,1].
|
||
macro_coeffs = iter([_mk_coef(c) for c in (0.5, 1.0, 1.5, 1.2, 0.8)])
|
||
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, side_effect=lambda *a, **k: next(macro_coeffs)),
|
||
):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num=None,
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
indices = [s["deficit_index"] for s in out["ranked_segments"]]
|
||
assert max(indices) == pytest.approx(1.0) # топ нормирован к 1.0
|
||
assert all(0.0 < i <= 1.0 for i in indices)
|
||
# DESC по pace.
|
||
assert indices == sorted(indices, reverse=True)
|
||
|
||
def test_no_velocity_yields_empty_plus_warning(self) -> None:
|
||
# base_pace None (нет §9.2 темпа) → НЕ фабрикуем сигнал, пустой ранкинг + warning.
|
||
with (
|
||
patch(_METRICS, return_value=_mk_metrics(None)),
|
||
patch(_GET_MACRO, return_value=[]),
|
||
patch(_HOLD, return_value={12: 18.0}),
|
||
):
|
||
out = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num=None,
|
||
horizon_months=12,
|
||
target_class=None,
|
||
)
|
||
assert out["mode"] == "demand_only"
|
||
assert out["ranked_segments"] == []
|
||
assert out["warnings"]
|
||
|
||
def test_five_default_room_buckets_ranked(self) -> None:
|
||
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,
|
||
)
|
||
buckets = {s["bucket"] for s in out["ranked_segments"]}
|
||
assert buckets == {"1-Студия", "2-1-к", "3-2-к", "4-3-к", "5-80+ м²"}
|
||
|
||
def test_overlay_validates_against_schema(self) -> None:
|
||
# Sanity: demand_only-выход проходит RecommendForecastOverlay-валидацию.
|
||
from app.schemas.recommend import RecommendForecastOverlay
|
||
|
||
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: dict[str, Any] = build_forecast_overlay(
|
||
MagicMock(),
|
||
district="Ленинский",
|
||
cad_num=None,
|
||
horizon_months=12,
|
||
target_class="Comfort",
|
||
)
|
||
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
|