lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(forecasting): demand-supply forecast engine (#980, 952-A §9.7)

Browse source 
Assemble merged §9.x services into per-(segment×horizon) DemandSupplyForecast:
demand (§9.2 pace × §9.4 norm × §9.5 macro, linear no-compound) vs supply
(open + phased hidden h/18 + future_online − absorbed, §9.3) → balance /
deficit_index[-1,+1] + §9.7 future competitors. Pure synthesis, no §9.x math
rebuilt; β counted EXACTLY ONCE (via §9.4 only; rate_sensitivity surfaced for
phrase only; §9.5 orthogonal). Supply net clamped ≥0; deficit_index None on
supply≤0 (no fake infinite deficit). confidence=MIN(components) capped medium;
advisory=True always. Deterministic, no LLM, no SQL, un-wired. 89 unit tests
(pure helpers + mocked orchestrator incl. adversarial β-once test), all green.
This commit is contained in:
Light1YT 2026-06-03 11:56:28 +05:00
parent 9758e21cbd
commit f3b04331f4
3 changed files with 1497 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
backend/app/services/forecasting/__init__.py
View file

@ -11,6 +11,8 @@
• rate_sensitivity (#951d) — §9.6 чувствительность продаж к key_rate (CORE, ADVISORY).
• macro_coefficient (#951e) — §9.5 макро-коэффициент (композитный множитель, ADVISORY).
• demand_normalization (#951f) — §9.4 нормализация спроса под смену режима ставки (ADVISORY).
• demand_supply_forecast (#952a) — §9.8 центральный движок: спрос (§9.4ק9.5) vs
предложение (§9.3) по горизонтам → баланс/индекс дефицита (СБОРКА, ADVISORY).
Источники данных:
• макро — таблица macro_indicator через reader site_finder/macro.py (reuse).
@ -24,6 +26,11 @@ from app.services.forecasting.demand_normalization import (
compute_demand_normalization,
normalization_factor,
)
from app.services.forecasting.demand_supply_forecast import (
DemandSupplyForecast,
compute_demand_supply_forecast,
hold_last_rate,
)
from app.services.forecasting.macro_coefficient import (
MacroCoefficient,
assemble_coefficient,
@ -61,6 +68,7 @@ from app.services.forecasting.sales_series import (
__all__ = [
"DemandNormalization",
"DemandSupplyForecast",
"MacroCoefficient",
"MonthlyMacro",
"RateSensitivity",
@ -71,6 +79,7 @@ __all__ = [
"build_sales_series",
"classify_regime",
"compute_demand_normalization",
"compute_demand_supply_forecast",
"compute_macro_coefficient",
"compute_rate_sensitivity",
"f_issuance",
@ -79,6 +88,7 @@ __all__ = [
"f_rate",
"fill_month_grid",
"get_monthly_macro",
"hold_last_rate",
"is_confounded_window",
"log_diff",
"macro_at_lag",

616
backend/app/services/forecasting/demand_supply_forecast.py Normal file
View file

@ -0,0 +1,616 @@
"""§9.4ק9.5 спрос vs §9.3 предложение — центральный прогнозный движок (ТЗ §9.8).
#952 (Site Finder v2 / «GG-форсайт» ТЗ §9.8), EPIC 8 «Центральный прогнозный
движок». Это **СБОРОЧНЫЙ слой**: он НЕ пересобирает §9.x-математику, а синтезирует
уже-смерженные сервисы в per-(сегмент × горизонт) баланс СПРОС↔ПРЕДЛОЖЕНИЕ:
• СПРОС = наблюдаемый темп (§9.2 unit_velocity) × нормализация под режим ставки
(§9.4 compute_demand_normalization) × макро-режим (§9.5
compute_macro_coefficient), спроектированный линейно на горизонт.
• ПРЕДЛОЖЕНИЕ = открытый сток + фазированный скрытый запас + будущий слой в
горизонте − поглощённое спросом (§9.3 compute_future_supply_pressure).
• БАЛАНС / индекс дефицита = знаковое насыщающее преобразование отношения
спрос/предложение в [−1,+1] (+1 = сильный дефицит «хорошо» / −1 =
сильная затоварка «риск»).
• БУДУЩИЕ КОНКУРЕНТЫ (§9.7) — top-N по relevance_weight на горизонт.
Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (чистая арифметика + reuse §9.x; своего SQL НЕТ).
ADVISORY-СТАТУС: движок СОВЕТУЮЩИЙ и НЕ подключён ни к одному production-эндпоинту
(как §9.4/§9.5/§9.6, все advisory до бэктеста #951 / валидации #951). `advisory`
поле ВСЕГДА True; итоговый confidence ЖЁСТКО ограничен сверху 'medium' (синтез не
надёжнее непровалидированных компонентов). Цифры — для explainability/прототипа.
──────────────────────────────────────────────────────────────────────────────
⚠️ ГЛАВНОЕ ПРАВИЛО КОРРЕКТНОСТИ — β УЧТЁН РОВНО ОДИН РАЗ:
• §9.4 `compute_demand_normalization` УЖЕ внутри вызывает §9.6
`compute_rate_sensitivity` и применяет β как exp(β·Δrate). Поэтому СПРОС
проходит через `compute_demand_normalization` ТОЛЬКО — мы НИГДЕ не домножаем
`rate_sensitivity.beta`/`x_pct` в число спроса (это было бы двойным учётом β).
• `compute_rate_sensitivity` здесь вызывается ИСКЛЮЧИТЕЛЬНО ради explainability —
чтобы вынести `.phrase` (и связанные `x_pct`/`y_lag_months`) в вывод. НИКАКОЙ
арифметики над её β/x_pct в расчёте спроса.
• §9.5 `macro_coefficient` ОРТОГОНАЛЕН β: он про каналы issuance/overdue/
mortgage-rate (макро-режим), а НЕ про эластичность к Δ ключевой ставки.
Поэтому `§9.4 × §9.5` — это два РАЗНЫХ множителя, НЕ двойной учёт.
ГОРИЗОНТ-ПРОЕКЦИЯ ЛИНЕЙНА (документируем): projected_demand = demand_per_mo × h,
БЕЗ компаундинга. Это сознательно: помесячный темп уже нормализован под будущий
режим (§9.4) и макро (§9.5); накручивать сверху сложный процент роста на
непровалидированном движке = ложная точность. Линейная проекция интерпретируема
(«столько ед. поглотит рынок за h мес при текущем нормализованном темпе») и
зеркалит дисциплину линейного clamp future_supply._saturating_index.
Graceful-on-thin-data (дух future_supply / market_metrics): любой тонкий вход →
соответствующее поле None (НИКОГДА 0-как-заглушка), НИКОГДА не crash, НИКОГДА
деления на ноль. supply ≤ 0 / None → balance_ratio=None, deficit_index=None,
confidence занижен (не выдумываем «бесконечный дефицит»). Детерминированно.
"""
from __future__ import annotations
import logging
import math
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any, Literal
from sqlalchemy.orm import Session
from app.schemas.parcel import CompetitorsRequest
from app.services.forecasting.demand_normalization import compute_demand_normalization
from app.services.forecasting.macro_coefficient import compute_macro_coefficient
from app.services.forecasting.macro_series import MonthlyMacro, get_monthly_macro
from app.services.forecasting.rate_sensitivity import compute_rate_sensitivity
from app.services.forecasting.sales_series import SegmentSpec
from app.services.site_finder.competitors import get_competitors
from app.services.site_finder.future_supply import compute_future_supply_pressure
from app.services.site_finder.market_metrics import compute_market_metrics
logger = logging.getLogger(__name__)
Confidence = Literal["high", "medium", "low"]
# ── Named-константы ───────────────────────────────────────────────────────────
# Горизонты прогноза по умолчанию (мес). 6/12/18/24 — зеркало §9.x горизонт-сетки
# (competitors.horizon_months / future_supply.horizon_months): полгода…2 года —
# окно, в котором перспективный объект реально выходит на рынок и конкурирует.
_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)
# Фазированный выход скрытого запаса (Layer2 ПД) — за сколько месяцев он полностью
# «созревает» в конкурирующее предложение. 18 мес ≈ типовой цикл от запаса ПД до
# вывода корпуса на продажу в ЕКБ. Доля выхода = clamp(h / _HIDDEN_RELEASE_MONTHS,
# 0, 1): на горизонте 6 мес вышла ~⅓ скрытого, 12 мес ~⅔, ≥18 мес — весь. Так
# скрытый слой давит ПОСТЕПЕННО (не весь сразу на ближнем горизонте и не игнор —
# в духе future_supply._horizon_weight, но непрерывно по доле, а не по дате).
_HIDDEN_RELEASE_MONTHS: float = 18.0
# Насыщение индекса дефицита: при balance_ratio = _DEFICIT_RATIO_SATURATION (спрос
# вдвое выше предложения) индекс достигает +1 (сильный дефицит); при обратном
# (предложение вдвое выше спроса, ratio = 1/2) → −1 (сильная затоварка). Знаковое
# насыщающее преобразование log-отношения (симметрично вокруг ratio=1 → 0): берём
# log(ratio)/log(saturation), clamp в [−1,+1]. 2.0 = «двукратный перекос — уже
# экстремум» (зеркало духа future_supply._PRESSURE_SATURATION линейного clamp, но
# здесь лог-шкала, т.к. отношение мультипликативно: ×2 и ÷2 симметричны). Tunable.
_DEFICIT_RATIO_SATURATION: float = 2.0
# Сколько будущих конкурентов (§9.7) выносим в вывод (top-N по relevance_weight).
_TOP_COMPETITORS: int = 5
# Радиус поиска конкурентов (км) для §9.7 — дефолт CompetitorsRequest (1 км =
# «прямые соседи по локации»). Держим явной константой для детерминизма вызова.
_COMPETITORS_RADIUS_KM: float = 1.0
# Жёсткий потолок итогового confidence: движок advisory до бэктеста #951, поэтому
# даже при всех 'high'-компонентах синтез не объявляем надёжнее 'medium'.
_CONFIDENCE_CAP: Confidence = "medium"
# Порядок уверенности для MIN-агрегации (хуже = ниже). Зеркало future_supply.
_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}
_RANK_TO_CONFIDENCE: dict[int, Confidence] = {0: "low", 1: "medium", 2: "high"}
@dataclass(frozen=True)
class DemandSupplyForecast:
"""Per-(сегмент × горизонт) баланс СПРОС↔ПРЕДЛОЖЕНИЕ (ТЗ §9.8, считается на лету).
Все величины детерминированы. Любое числовое поле = None при недостатке данных
(НИКОГДА 0-как-заглушка). `advisory` ВСЕГДА True (движок не для production-решений
до бэктеста #951). `deficit_index` ∈ [−1,+1] когда задан: +1 = сильный дефицит
(мало предложения под спрос — «хорошо» для девелопера), −1 = сильная затоварка
(предложения больше спроса — «риск»); None при неизмеримом предложении.
`rate_sensitivity_phrase` — ТОЛЬКО explainability (β НЕ участвует в арифметике
спроса дважды — он уже учтён внутри demand_norm_coefficient §9.4).
"""
# ── Контекст ──────────────────────────────────────────────────────────────
segment: dict[str, str | None]
horizon_months: int
# ── Спрос (§9.2 темп × §9.4 норм × §9.5 макро, линейно на горизонт) ────────
base_pace_units_per_mo: float | None # §9.2 observed unit_velocity (ед./мес)
demand_norm_coefficient: float | None # §9.4 множитель (β внутри — учтён 1 раз)
macro_coefficient: float | None # §9.5 множитель (ортогонален β)
projected_demand_units: float | None # base_pace × §9.4 × §9.5 × h (линейно)
# ── Предложение (§9.3 слои, фазированный hidden, за вычетом поглощённого) ───
open_units: int # Σ Layer1 (в продаже) — контекст
hidden_release_units: float # Σ Layer2 × фаза выхода на горизонте
future_online_units: float # Σ Layer3, взвешенный по попаданию в горизонт
projected_supply_units: float # open + hidden_release + future − absorbed, ≥0
# ── Баланс / индекс дефицита ───────────────────────────────────────────────
balance_units: float | None # demand − supply (>0 дефицит / <0 затоварка)
balance_ratio: float | None # demand / supply (None если supply ≤ 0)
deficit_index: float | None # знаковое насыщение balance_ratio → [−1,+1]
# ── Explainability ─────────────────────────────────────────────────────────
rate_future: float | None # ставка сценария на горизонте (вход §9.4)
rate_sensitivity_phrase: str | None # §9.6 фраза (НЕ арифметика — explain-only)
future_competitors: list[dict[str, Any]] # §9.7 top-N по relevance_weight
advisory: bool # ВСЕГДА True (движок не для production-решений)
confidence: Confidence # MIN(компоненты), жёстко ≤ _CONFIDENCE_CAP
def as_dict(self) -> dict[str, Any]:
return {
"segment": dict(self.segment),
"horizon_months": self.horizon_months,
"base_pace_units_per_mo": _round_or_none(self.base_pace_units_per_mo, 2),
"demand_norm_coefficient": _round_or_none(self.demand_norm_coefficient, 4),
"macro_coefficient": _round_or_none(self.macro_coefficient, 4),
"projected_demand_units": _round_or_none(self.projected_demand_units, 1),
"open_units": self.open_units,
"hidden_release_units": _round_or_none(self.hidden_release_units, 1),
"future_online_units": _round_or_none(self.future_online_units, 1),
"projected_supply_units": _round_or_none(self.projected_supply_units, 1),
"balance_units": _round_or_none(self.balance_units, 1),
"balance_ratio": _round_or_none(self.balance_ratio, 3),
"deficit_index": _round_or_none(self.deficit_index, 3),
"rate_future": _round_or_none(self.rate_future, 2),
"rate_sensitivity_phrase": self.rate_sensitivity_phrase,
"future_competitors": list(self.future_competitors),
"advisory": self.advisory,
"confidence": self.confidence,
}
def _round_or_none(value: float | None, digits: int) -> float | None:
return round(value, digits) if value is not None else None
# ──────────────────────────────────────────────────────────────────────────────
# Pure-арифметика — без БД, полностью юнит-тестируемо.
# Каждая функция graceful: тонкий/нулевой вход → None/нейтраль (не crash, не /0).
# ──────────────────────────────────────────────────────────────────────────────
def _hidden_release_fraction(
horizon_months: int, *, release_months: float = _HIDDEN_RELEASE_MONTHS
) -> float:
"""Доля скрытого запаса (Layer2), вышедшая в предложение к горизонту ∈ [0,1].
clamp(h / release_months, 0, 1): фазированный выход скрытого ПД в конкурирующее
предложение. На горизонте 6 мес из 18-месячного цикла вышла треть, 12 → две
трети, ≥18 → весь. Так скрытый слой давит ПОСТЕПЕННО (не весь сразу). PURE.
Args:
horizon_months: горизонт прогноза (мес); ≤0 → 0.0 (ничего не вышло).
release_months: полный цикл созревания скрытого запаса (по умолчанию
_HIDDEN_RELEASE_MONTHS); ≤0 → деградация: любой h>0 → 1.0 (весь сразу).
Returns:
Доля в [0,1].
"""
if horizon_months <= 0:
return 0.0
if release_months <= 0:
return 1.0
return max(0.0, min(1.0, horizon_months / release_months))
def _project_demand(
base_pace: float | None,
norm_coefficient: float | None,
macro_coefficient: float | None,
horizon_months: int,
) -> float | None:
"""Спроектированный спрос = base_pace × §9.4 × §9.5 × h (ЛИНЕЙНО, без компаунда).
base_pace None → None (НЕ 0: «нет данных о рынке» ≠ «продажи 0»). Коэффициенты
None трактуем как 1.0 (нейтраль — компонент не скорректировал темп, но темп
известен). β учтён РОВНО ОДИН РАЗ внутри norm_coefficient (§9.4) — здесь его
больше НЕ домножаем. Линейность сознательна (см. module docstring). PURE.
Args:
base_pace: §9.2 наблюдаемый темп (ед./мес); None = нет выборки.
norm_coefficient: §9.4 множитель нормализации (β уже внутри); None → 1.0.
macro_coefficient: §9.5 макро-множитель (ортогонален β); None → 1.0.
horizon_months: горизонт проекции (мес); ≤0 → 0.0 спроса.
Returns:
Спроектированные ед. спроса или None (нет base_pace).
"""
if base_pace is None:
return None
if horizon_months <= 0:
return 0.0
norm = norm_coefficient if norm_coefficient is not None else 1.0
macro = macro_coefficient if macro_coefficient is not None else 1.0
demand_per_mo = base_pace * norm * macro
return demand_per_mo * float(horizon_months)
def _project_supply(
open_units: float,
hidden_release_units: float,
future_online_units: float,
projected_demand_units: float | None,
) -> float:
"""Чистое конкурирующее предложение = валовое − поглощённое спросом, clamp ≥0.
Валовое = open + hidden_release + future_online (всё, что выйдет/доступно на
горизонте). Поглощённое (absorbed) = min(валовое, projected_demand) — рынок
«съест» столько, сколько есть спроса, но не больше доступного объёма. Чистое =
валовое − absorbed (остаток, который РЕАЛЬНО конкурирует с нашим объектом).
Спрос None трактуем как 0 поглощения (нет данных о спросе → ничего не вычитаем
→ консервативно показываем всё валовое как конкурирующее). clamp ≥0. PURE.
Args:
open_units: Σ Layer1 (в продаже).
hidden_release_units: Σ Layer2 × фаза выхода на горизонте.
future_online_units: Σ Layer3, взвешенный по горизонту.
projected_demand_units: спроектированный спрос (None → 0 поглощения).
Returns:
Чистое конкурирующее предложение (≥0).
"""
gross = open_units + hidden_release_units + future_online_units
demand = projected_demand_units if projected_demand_units is not None else 0.0
absorbed = min(gross, max(0.0, demand))
return max(0.0, gross - absorbed)
def _balance(
projected_demand_units: float | None, projected_supply_units: float | None
) -> float | None:
"""balance_units = спрос − предложение (>0 дефицит / <0 затоварка). PURE.
Любой вход None → None (нечего сравнивать). >0 = спроса больше, чем
конкурирующего предложения (дефицит — «хорошо» для девелопера); <0 = затоварка.
"""
if projected_demand_units is None or projected_supply_units is None:
return None
return projected_demand_units - projected_supply_units
def _balance_ratio(
projected_demand_units: float | None, projected_supply_units: float | None
) -> float | None:
"""balance_ratio = спрос / предложение. PURE.
supply ≤ 0 / None ИЛИ demand None → None (НЕ ∞: «предложение исчерпано» не
отличить от «нет данных», поэтому честно None, а не выдуманный бесконечный
дефицит — индекс тогда тоже None, confidence занижается выше по стеку).
"""
if projected_demand_units is None or projected_supply_units is None:
return None
if projected_supply_units <= 0:
return None
return projected_demand_units / projected_supply_units
def _deficit_index(
balance_ratio: float | None, *, saturation: float = _DEFICIT_RATIO_SATURATION
) -> float | None:
"""Знаковое насыщение balance_ratio → индекс дефицита ∈ [−1,+1]. PURE.
Лог-шкала (отношение мультипликативно — ×2 и ÷2 симметричны вокруг 1.0):
index = clamp(log(ratio) / log(saturation), −1, +1).
• ratio = 1.0 (спрос = предложение) → log(1)=0 → index 0 (баланс).
• ratio = saturation (спрос вдвое > предложения) → +1 (сильный дефицит).
• ratio = 1/saturation (предложение вдвое > спроса) → −1 (сильная затоварка).
None / непозитивный ratio → None (неизмеримо). saturation ≤1 → деградация:
знак ratio−1 (без насыщающей шкалы). Монотонно неубывающее по ratio. PURE.
Args:
balance_ratio: спрос/предложение (None → None).
saturation: ratio, при котором индекс достигает ±1 (по умолч.
_DEFICIT_RATIO_SATURATION).
Returns:
Индекс в [−1,+1] или None.
"""
if balance_ratio is None or balance_ratio <= 0:
return None
if saturation <= 1.0:
# Деградация без падения: нет осмысленной шкалы → только знак перекоса.
if balance_ratio > 1.0:
return 1.0
if balance_ratio < 1.0:
return -1.0
return 0.0
raw = math.log(balance_ratio) / math.log(saturation)
return max(-1.0, min(1.0, raw))
def _min_confidence(values: Sequence[Confidence | None]) -> Confidence:
"""Итоговая уверенность = MIN компонентных (худшая тянет вниз). Зеркало vocab.
Тонкий любой компонент (market_metrics / §9.4 / §9.5 / future_supply) честно
роняет общий confidence. None в списке игнорируем (компонент без сигнала).
Пустой/весь-None вход → 'low'. Только whitelisted 'high|medium|low'. PURE.
"""
ranks = [_CONFIDENCE_RANK[v] for v in values if v is not None]
if not ranks:
return "low"
return _RANK_TO_CONFIDENCE[min(ranks)]
def _cap_confidence(confidence: Confidence, *, cap: Confidence = _CONFIDENCE_CAP) -> Confidence:
"""Ограничить confidence сверху (advisory-движок не надёжнее cap). PURE.
Берём ранг-минимум(confidence, cap): high+cap-medium → medium; low → low.
"""
capped_rank = min(_CONFIDENCE_RANK[confidence], _CONFIDENCE_RANK[cap])
return _RANK_TO_CONFIDENCE[capped_rank]
def hold_last_rate(macro: list[MonthlyMacro], horizons: Sequence[int]) -> dict[int, float | None]:
"""Дефолтный rate-path: последняя известная key_rate, удержанная плоско. PURE.
Сценарий «ставка не меняется»: берём самую свежую НЕпустую key_rate из макро-
ряда и присваиваем её КАЖДОМУ горизонту. Нет ни одной точки key_rate → None на
всех горизонтах (graceful: §9.4 тогда деградирует к нейтрали внутри себя).
Args:
macro: monthly макро-ряд (§9.5/§9.6 PR2 get_monthly_macro).
horizons: горизонты (мес), под которые строим path.
Returns:
{horizon: rate}; rate = последняя key_rate (или None) на каждом горизонте.
"""
last_rate: float | None = None
for m in macro: # ряд ASC по month → последняя непустая = самая свежая
if m.key_rate is not None:
last_rate = m.key_rate
return {h: last_rate for h in horizons}
# ──────────────────────────────────────────────────────────────────────────────
# DB-оркестратор — тонкий, graceful. Pure-логика выше тестируется без него.
# ──────────────────────────────────────────────────────────────────────────────
def compute_demand_supply_forecast(
db: Session,
*,
spec: SegmentSpec,
district: str | None,
cad_num: str,
horizons: Sequence[int] = _DEFAULT_HORIZONS,
rate_path: dict[int, float] | None = None,
premise_kind: str = "квартира",
) -> list[DemandSupplyForecast]:
"""Собрать per-горизонт прогноз баланса СПРОС↔ПРЕДЛОЖЕНИЕ (ТЗ §9.8) для сегмента.
ADVISORY (все §9.x-компоненты advisory до бэктеста #951) — НЕ подключать в
production-эндпоинт. СБОРКА, НЕ пересчёт §9.x: вызывает уже-смерженные сервисы.
⚠️ β УЧТЁН РОВНО ОДИН РАЗ: спрос идёт через §9.4 compute_demand_normalization
(она сама применяет β внутри); §9.6 compute_rate_sensitivity вызывается ТОЛЬКО
ради explainability-фразы — её β/x_pct в арифметику спроса НЕ входят.
Один раз на вызов:
• macro = get_monthly_macro(db) — для дефолтного rate-path (hold_last_rate).
• base_pace = compute_market_metrics(...).unit_velocity (§9.2 ед./мес).
• §9.5 compute_macro_coefficient (макро-режим, ортогонален β).
• §9.6 compute_rate_sensitivity (ТОЛЬКО фраза для explain).
На каждый горизонт h:
• rate_future = rate_path[h] (или hold_last_rate).
• §9.4 compute_demand_normalization(rate_future=rate_future) → norm (β внутри).
• projected_demand = base_pace × norm × §9.5 × h (линейно, без компаунда).
• §9.3 compute_future_supply_pressure(horizon_months=h) → open/hidden/future;
hidden_release = hidden × _hidden_release_fraction(h); чистое предложение =
open + hidden_release + future − поглощённое спросом (clamp ≥0).
• balance / ratio / знаковый deficit_index.
• §9.7 future_competitors = top-N get_competitors(horizon_months=h).
• confidence = MIN(market, §9.4, §9.5, future_supply) ≤ 'medium'.
Graceful: любой тонкий вход → поле None, НЕ crash. supply ≤0/None → ratio/index
None. Конкуренты недоступны (нет геометрии cad_num) → []. Детерминированно.
Args:
db: SQLAlchemy sync Session.
spec: целевой сегмент рынка (любой subset осей).
district: район для §9.3 supply + §9.2 metrics (None → ЕКБ-wide).
cad_num: кадастровый номер участка — вход для §9.7 конкурентов.
horizons: горизонты прогноза (мес; по умолчанию _DEFAULT_HORIZONS).
rate_path: сценарный {horizon: key_rate}; None → hold_last_rate (плоско).
premise_kind: тип помещения (по умолчанию 'квартира').
Returns:
Список DemandSupplyForecast — по одному на горизонт (всегда; тонко → None-поля).
"""
horizon_list = list(horizons)
segment = spec.as_dict()
profile = _segment_profile(spec)
# ── Один раз на вызов: макро-ряд + дефолтный rate-path ─────────────────────
macro = get_monthly_macro(db)
effective_rate_path = (
dict(rate_path) if rate_path is not None else hold_last_rate(macro, horizon_list)
)
# ── Один раз: §9.2 наблюдаемый темп (base_pace) ────────────────────────────
metrics = compute_market_metrics(db, district=district, premise_kind=premise_kind)
base_pace = metrics.unit_velocity
# ── Один раз: §9.5 макро-коэффициент (ортогонален β) ───────────────────────
macro_coef = compute_macro_coefficient(db, segment_profile=profile)
# ── Один раз: §9.6 чувствительность — ТОЛЬКО для explain-фразы (НЕ арифметика)
sensitivity = compute_rate_sensitivity(db, spec=spec)
out: list[DemandSupplyForecast] = []
for h in horizon_list:
out.append(
_forecast_for_horizon(
db,
spec=spec,
segment=segment,
district=district,
cad_num=cad_num,
horizon=h,
rate_future=effective_rate_path.get(h),
base_pace=base_pace,
market_confidence=metrics.confidence,
macro_coef=macro_coef,
sensitivity_phrase=sensitivity.phrase,
premise_kind=premise_kind,
)
)
return out
def _forecast_for_horizon(
db: Session,
*,
spec: SegmentSpec,
segment: dict[str, str | None],
district: str | None,
cad_num: str,
horizon: int,
rate_future: float | None,
base_pace: float | None,
market_confidence: Confidence,
macro_coef: Any,
sensitivity_phrase: str | None,
premise_kind: str,
) -> DemandSupplyForecast:
"""Собрать прогноз для ОДНОГО горизонта (тонкий — pure-логика выше). Graceful."""
# ── СПРОС: §9.4 (β внутри — ОДИН раз) → линейная проекция × §9.5 ───────────
# rate_future None → §9.4 деградирует к нейтрали внутри себя (передаём 0.0 как
# placeholder ставки, но при норм-coef из low-conf β результат всё равно 1.0).
norm = compute_demand_normalization(
db, spec=spec, rate_future=rate_future if rate_future is not None else 0.0
)
norm_coefficient = norm.coefficient if rate_future is not None else None
macro_coefficient = macro_coef.coefficient
projected_demand = _project_demand(base_pace, norm_coefficient, macro_coefficient, horizon)
# ── ПРЕДЛОЖЕНИЕ: §9.3 слои → фазированный hidden → чистое (− поглощённое) ───
fsp = compute_future_supply_pressure(
db, district=district, horizon_months=horizon, premise_kind=premise_kind
)
hidden_fraction = _hidden_release_fraction(horizon)
hidden_release = fsp.hidden_units * hidden_fraction
future_online = fsp.future_units_by_horizon
projected_supply = _project_supply(
fsp.open_units, hidden_release, future_online, projected_demand
)
# ── БАЛАНС / индекс дефицита ────────────────────────────────────────────────
balance_units = _balance(projected_demand, projected_supply)
balance_ratio = _balance_ratio(projected_demand, projected_supply)
deficit_index = _deficit_index(balance_ratio)
# ── §9.7 будущие конкуренты (top-N по relevance_weight) ────────────────────
future_competitors = _future_competitors(db, cad_num=cad_num, horizon=horizon)
# ── confidence = MIN(компоненты) ≤ cap ─────────────────────────────────────
confidence = _cap_confidence(
_min_confidence([market_confidence, norm.confidence, macro_coef.confidence, fsp.confidence])
)
logger.info(
"demand_supply_forecast: segment=%s h=%d base_pace=%s norm=%s macro=%s "
"demand=%s supply=%.1f balance=%s ratio=%s deficit_index=%s confidence=%s",
segment,
horizon,
_round_or_none(base_pace, 2),
_round_or_none(norm_coefficient, 4),
_round_or_none(macro_coefficient, 4),
_round_or_none(projected_demand, 1),
projected_supply,
_round_or_none(balance_units, 1),
_round_or_none(balance_ratio, 3),
_round_or_none(deficit_index, 3),
confidence,
)
return DemandSupplyForecast(
segment=segment,
horizon_months=horizon,
base_pace_units_per_mo=base_pace,
demand_norm_coefficient=norm_coefficient,
macro_coefficient=macro_coefficient,
projected_demand_units=projected_demand,
open_units=fsp.open_units,
hidden_release_units=hidden_release,
future_online_units=future_online,
projected_supply_units=projected_supply,
balance_units=balance_units,
balance_ratio=balance_ratio,
deficit_index=deficit_index,
rate_future=rate_future,
rate_sensitivity_phrase=sensitivity_phrase,
future_competitors=future_competitors,
advisory=True,
confidence=confidence,
)
def _segment_profile(spec: SegmentSpec) -> dict[str, Any]:
"""Спроецировать SegmentSpec на профиль для §9.5 segment_steepness. PURE.
§9.5 segment_steepness читает obj_class / price_tier / room_bucket. Маппим
оси spec на ожидаемые ключи (price_bucket → price_tier). None-оси не кладём
(segment_steepness терпит отсутствие — нейтральная крутизна).
"""
profile: dict[str, Any] = {}
if spec.obj_class is not None:
profile["obj_class"] = spec.obj_class
if spec.room_bucket is not None:
profile["room_bucket"] = spec.room_bucket
if spec.price_bucket is not None:
profile["price_tier"] = spec.price_bucket
return profile
def _future_competitors(db: Session, *, cad_num: str, horizon: int) -> list[dict[str, Any]]:
"""§9.7 top-N будущих конкурентов по relevance_weight на горизонт. Graceful → [].
get_competitors сам horizon-aware (stage_at_horizon). Сортирует по
relevance_weight DESC, поэтому берём первые _TOP_COMPETITORS. Нет геометрии
участка (ValueError) / сбой → [] (не валим прогноз — конкуренты опциональны).
"""
try:
response = get_competitors(
db,
cad_num,
CompetitorsRequest(horizon_months=horizon, radius_km=_COMPETITORS_RADIUS_KM),
)
except Exception:
logger.exception(
"demand_supply_forecast: competitors lookup failed (cad_num=%s horizon=%d)",
cad_num,
horizon,
)
return []
return [_competitor_to_dict(c) for c in response.competitors[:_TOP_COMPETITORS]]
def _competitor_to_dict(competitor: Any) -> dict[str, Any]:
"""Свести Competitor в компактный explain-словарь (релевантные §9.7 поля). PURE-ish."""
return {
"obj_id": competitor.obj_id,
"comm_name": competitor.comm_name,
"obj_class": competitor.obj_class,
"distance_m": competitor.distance_m,
"flats_total": competitor.flats_total,
"velocity_per_month": competitor.velocity_per_month,
"relevance_weight": competitor.relevance_weight,
}

871
backend/tests/services/forecasting/test_demand_supply_forecast.py Normal file
View file

@ -0,0 +1,871 @@
"""Unit-тесты §9.8 центрального движка СПРОС↔ПРЕДЛОЖЕНИЕ (#952a, ADVISORY).
Чистые тесты — БЕЗ живой БД (чистая математика + мок всех §9.x compute_*):
• pure-арифметика: _hidden_release_fraction (6→⅓ / 12→⅔ / 18→1.0 / clamp),
_project_demand (paceק9.4ק9.5×h ЛИНЕЙНО; None pace → None; None-коэф → 1.0),
_project_supply (absorbed-clamp ≥0; None спрос → 0 поглощения),
_balance / _balance_ratio (знак + None при supply≤0),
_deficit_index (знаковое насыщение + clamp [−1,+1] + None), _min_confidence,
_cap_confidence (≤ medium), hold_last_rate (последняя key_rate плоско).
• compute_demand_supply_forecast через MagicMock-сессию + @patch всех reused
сервисов (зеркало test_future_supply): спрос = paceק9.4ק9.5×h ТОЧНО; β НЕ
учтён дважды (rate_sensitivity.beta/x_pct не влияет на projected_demand);
confidence = MIN ≤ medium; advisory ВСЕГДА True; один результат на горизонт;
graceful empty → None-поля; конкуренты-сбой → [].
Детерминированно, без LLM. Мокаем все §9.x compute_* + db (нет живой БД).
"""
from __future__ import annotations
import math
import os
from unittest.mock import MagicMock, patch
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import datetime as dt
from typing import Any, ClassVar
import pytest
from app.services.forecasting.demand_supply_forecast import (
_CONFIDENCE_CAP,
_DEFICIT_RATIO_SATURATION,
_HIDDEN_RELEASE_MONTHS,
_TOP_COMPETITORS,
DemandSupplyForecast,
_balance,
_balance_ratio,
_cap_confidence,
_deficit_index,
_hidden_release_fraction,
_min_confidence,
_project_demand,
_project_supply,
_round_or_none,
_segment_profile,
compute_demand_supply_forecast,
hold_last_rate,
)
from app.services.forecasting.macro_series import MonthlyMacro
from app.services.forecasting.sales_series import SegmentSpec
# Пути патча reused-сервисов (импортированы в модуль demand_supply_forecast).
_MACRO = "app.services.forecasting.demand_supply_forecast.get_monthly_macro"
_METRICS = "app.services.forecasting.demand_supply_forecast.compute_market_metrics"
_NORM = "app.services.forecasting.demand_supply_forecast.compute_demand_normalization"
_MACRO_COEF = "app.services.forecasting.demand_supply_forecast.compute_macro_coefficient"
_SENS = "app.services.forecasting.demand_supply_forecast.compute_rate_sensitivity"
_SUPPLY = "app.services.forecasting.demand_supply_forecast.compute_future_supply_pressure"
_COMPETITORS = "app.services.forecasting.demand_supply_forecast.get_competitors"
_ALLOWED_CONFIDENCE = {"high", "medium", "low"}
# ── pure: _hidden_release_fraction ────────────────────────────────────────────
class TestHiddenReleaseFraction:
def test_six_months_is_third(self) -> None:
# 6 / 18 = 0.333… (≈⅓ скрытого запаса вышло к 6 мес).
assert _hidden_release_fraction(6) == pytest.approx(1.0 / 3.0)
def test_twelve_months_is_two_thirds(self) -> None:
assert _hidden_release_fraction(12) == pytest.approx(2.0 / 3.0)
def test_eighteen_months_is_full(self) -> None:
# ровно цикл созревания → весь скрытый запас вышел (1.0).
assert _hidden_release_fraction(18) == pytest.approx(1.0)
def test_beyond_cycle_clamps_to_one(self) -> None:
assert _hidden_release_fraction(24) == 1.0
assert _hidden_release_fraction(36) == 1.0
def test_zero_or_negative_horizon_is_zero(self) -> None:
assert _hidden_release_fraction(0) == 0.0
assert _hidden_release_fraction(-5) == 0.0
def test_uses_named_constant(self) -> None:
# На горизонте = _HIDDEN_RELEASE_MONTHS → ровно 1.0 (привязка к константе).
assert _hidden_release_fraction(int(_HIDDEN_RELEASE_MONTHS)) == pytest.approx(1.0)
def test_degraded_zero_release_months(self) -> None:
# release_months ≤0 → деградация: любой h>0 → весь сразу (1.0), не /0.
assert _hidden_release_fraction(6, release_months=0.0) == 1.0
assert _hidden_release_fraction(0, release_months=0.0) == 0.0
# ── pure: _project_demand (ЛИНЕЙНО, β внутри §9.4 — один раз) ──────────────────
class TestProjectDemand:
def test_linear_product(self) -> None:
# 10 ед/мес × 0.8 (§9.4) × 1.1 (§9.5) × 12 мес = 105.6 (ЛИНЕЙНО, без компаунда).
assert _project_demand(10.0, 0.8, 1.1, 12) == pytest.approx(10.0 * 0.8 * 1.1 * 12)
def test_linear_not_compounded(self) -> None:
# Двойной горизонт → ровно двойной спрос (линейность, НЕ (1+r)^h).
d12 = _project_demand(5.0, 0.9, 1.0, 12)
d24 = _project_demand(5.0, 0.9, 1.0, 24)
assert d12 is not None and d24 is not None
assert d24 == pytest.approx(2.0 * d12)
def test_none_pace_returns_none(self) -> None:
# нет наблюдаемого темпа → спрос None (НЕ 0: «нет данных» ≠ «продажи 0»).
assert _project_demand(None, 0.8, 1.1, 12) is None
def test_none_coefficients_treated_as_neutral(self) -> None:
# коэф None → 1.0 (компонент не скорректировал, но темп известен).
assert _project_demand(10.0, None, None, 12) == pytest.approx(120.0)
assert _project_demand(10.0, 0.5, None, 12) == pytest.approx(60.0)
def test_zero_horizon_zero_demand(self) -> None:
assert _project_demand(10.0, 0.8, 1.1, 0) == 0.0
def test_zero_pace_is_zero_not_none(self) -> None:
# темп измерим и равен 0 → спрос 0 (валидно, НЕ None).
assert _project_demand(0.0, 0.8, 1.1, 12) == 0.0
# ── pure: _project_supply (absorbed-clamp ≥0) ─────────────────────────────────
class TestProjectSupply:
def test_net_supply_after_absorption(self) -> None:
# валовое = 100+50+30 = 180; спрос 80 → absorbed 80 → чистое 100.
assert _project_supply(100.0, 50.0, 30.0, 80.0) == pytest.approx(100.0)
def test_demand_exceeds_supply_clamps_to_zero(self) -> None:
# спрос 500 > валовое 180 → absorbed = всё 180 → чистое 0 (clamp ≥0, не отриц).
assert _project_supply(100.0, 50.0, 30.0, 500.0) == 0.0
def test_none_demand_no_absorption(self) -> None:
# спрос None → 0 поглощения → всё валовое конкурирует (консервативно).
assert _project_supply(100.0, 50.0, 30.0, None) == pytest.approx(180.0)
def test_zero_demand_no_absorption(self) -> None:
assert _project_supply(100.0, 50.0, 30.0, 0.0) == pytest.approx(180.0)
def test_negative_demand_treated_as_zero(self) -> None:
# артефактно-отрицательный спрос не «добавляет» предложение (max(0, demand)).
assert _project_supply(100.0, 0.0, 0.0, -50.0) == pytest.approx(100.0)
def test_empty_supply_is_zero(self) -> None:
assert _project_supply(0.0, 0.0, 0.0, 80.0) == 0.0
def test_result_never_negative(self) -> None:
for demand in [0.0, 10.0, 50.0, 100.0, 250.0, 1000.0]:
assert _project_supply(40.0, 30.0, 20.0, demand) >= 0.0
# ── pure: _balance ────────────────────────────────────────────────────────────
class TestBalance:
def test_deficit_positive(self) -> None:
# спрос 200 > предложение 120 → +80 (дефицит — «хорошо»).
assert _balance(200.0, 120.0) == pytest.approx(80.0)
def test_oversupply_negative(self) -> None:
# спрос 80 < предложение 120 → −40 (затоварка — «риск»).
assert _balance(80.0, 120.0) == pytest.approx(-40.0)
def test_none_demand_returns_none(self) -> None:
assert _balance(None, 120.0) is None
def test_none_supply_returns_none(self) -> None:
assert _balance(200.0, None) is None
# ── pure: _balance_ratio (None при supply ≤0) ─────────────────────────────────
class TestBalanceRatio:
def test_basic_ratio(self) -> None:
assert _balance_ratio(200.0, 100.0) == pytest.approx(2.0)
def test_zero_supply_returns_none(self) -> None:
# предложение исчерпано → None (НЕ ∞: не отличить от «нет данных»).
assert _balance_ratio(200.0, 0.0) is None
def test_negative_supply_returns_none(self) -> None:
assert _balance_ratio(200.0, -5.0) is None
def test_none_demand_returns_none(self) -> None:
assert _balance_ratio(None, 100.0) is None
def test_none_supply_returns_none(self) -> None:
assert _balance_ratio(200.0, None) is None
# ── pure: _deficit_index (знаковое насыщение [−1,+1]) ─────────────────────────
class TestDeficitIndex:
def test_balance_is_zero(self) -> None:
# спрос = предложение (ratio 1.0) → индекс 0 (баланс).
assert _deficit_index(1.0) == pytest.approx(0.0)
def test_saturation_ratio_is_plus_one(self) -> None:
# ratio = saturation (спрос вдвое > предложения) → +1 (сильный дефицит).
assert _deficit_index(_DEFICIT_RATIO_SATURATION) == pytest.approx(1.0)
def test_inverse_saturation_is_minus_one(self) -> None:
# ratio = 1/saturation (предложение вдвое > спроса) → −1 (сильная затоварка).
assert _deficit_index(1.0 / _DEFICIT_RATIO_SATURATION) == pytest.approx(-1.0)
def test_beyond_saturation_clamps_to_plus_one(self) -> None:
assert _deficit_index(_DEFICIT_RATIO_SATURATION * 5.0) == 1.0
def test_far_below_clamps_to_minus_one(self) -> None:
assert _deficit_index(1.0 / (_DEFICIT_RATIO_SATURATION * 5.0)) == -1.0
def test_symmetric_around_balance(self) -> None:
# ×r и ÷r симметричны вокруг 0 (лог-шкала отношения).
r = 1.5
assert _deficit_index(r) == pytest.approx(-_deficit_index(1.0 / r))
def test_none_ratio_returns_none(self) -> None:
assert _deficit_index(None) is None
def test_nonpositive_ratio_returns_none(self) -> None:
assert _deficit_index(0.0) is None
assert _deficit_index(-1.0) is None
def test_monotonic_non_decreasing(self) -> None:
prev = -2.0
for ratio in [0.1, 0.25, 0.5, 0.75, 1.0, 1.5, 2.0, 4.0, 10.0]:
idx = _deficit_index(ratio)
assert idx is not None
assert -1.0 <= idx <= 1.0
assert idx >= prev
prev = idx
def test_degraded_saturation_le_one(self) -> None:
# saturation ≤1 → деградация: только знак перекоса (не /0 в log).
assert _deficit_index(2.0, saturation=1.0) == 1.0
assert _deficit_index(0.5, saturation=1.0) == -1.0
assert _deficit_index(1.0, saturation=1.0) == 0.0
# ── pure: _min_confidence / _cap_confidence ───────────────────────────────────
class TestMinConfidence:
def test_all_high(self) -> None:
assert _min_confidence(["high", "high"]) == "high"
def test_low_drags_down(self) -> None:
assert _min_confidence(["high", "medium", "low"]) == "low"
def test_medium_floor(self) -> None:
assert _min_confidence(["high", "medium"]) == "medium"
def test_none_ignored(self) -> None:
assert _min_confidence(["high", None, "medium"]) == "medium"
def test_all_none_low(self) -> None:
assert _min_confidence([None, None]) == "low"
def test_empty_low(self) -> None:
assert _min_confidence([]) == "low"
class TestCapConfidence:
def test_high_capped_to_medium(self) -> None:
# advisory-движок: даже all-high → не выше medium.
assert _cap_confidence("high") == "medium"
assert _CONFIDENCE_CAP == "medium"
def test_medium_stays_medium(self) -> None:
assert _cap_confidence("medium") == "medium"
def test_low_stays_low(self) -> None:
assert _cap_confidence("low") == "low"
def test_custom_cap(self) -> None:
assert _cap_confidence("high", cap="low") == "low"
# ── pure: hold_last_rate ──────────────────────────────────────────────────────
def _macro_pt(month: dt.date, key_rate: float | None) -> MonthlyMacro:
return MonthlyMacro(
month=month,
key_rate=key_rate,
mortgage_rate_weighted=None,
mortgage_issued_count=None,
mortgage_issued_volume=None,
mortgage_debt=None,
mortgage_overdue=None,
)
class TestHoldLastRate:
def test_holds_latest_known_rate_flat(self) -> None:
macro = [
_macro_pt(dt.date(2025, 1, 1), 16.0),
_macro_pt(dt.date(2025, 2, 1), 18.0),
_macro_pt(dt.date(2025, 3, 1), 21.0),
]
path = hold_last_rate(macro, [6, 12, 24])
assert path == {6: 21.0, 12: 21.0, 24: 21.0}
def test_ignores_trailing_none(self) -> None:
# последняя НЕпустая key_rate (трейлинг None игнорируем).
macro = [
_macro_pt(dt.date(2025, 1, 1), 16.0),
_macro_pt(dt.date(2025, 2, 1), None),
]
assert hold_last_rate(macro, [12]) == {12: 16.0}
def test_all_none_yields_none(self) -> None:
macro = [_macro_pt(dt.date(2025, 1, 1), None)]
assert hold_last_rate(macro, [6, 12]) == {6: None, 12: None}
def test_empty_macro_yields_none(self) -> None:
assert hold_last_rate([], [6, 12]) == {6: None, 12: None}
# ── pure: _segment_profile ────────────────────────────────────────────────────
class TestSegmentProfile:
def test_maps_axes(self) -> None:
spec = SegmentSpec(obj_class="бизнес", room_bucket="3", price_bucket="премиум")
prof = _segment_profile(spec)
assert prof["obj_class"] == "бизнес"
assert prof["room_bucket"] == "3"
assert prof["price_tier"] == "премиум" # price_bucket → price_tier
def test_omits_none_axes(self) -> None:
prof = _segment_profile(SegmentSpec())
assert prof == {}
# ── pure: _round_or_none ──────────────────────────────────────────────────────
class TestRoundOrNone:
def test_rounds(self) -> None:
assert _round_or_none(1.23456, 3) == 1.235
def test_none_passthrough(self) -> None:
assert _round_or_none(None, 3) is None
# ── DemandSupplyForecast.as_dict ──────────────────────────────────────────────
def _make_forecast(**over: object) -> DemandSupplyForecast:
base: dict[str, object] = {
"segment": {
"obj_class": "комфорт",
"room_bucket": None,
"district": "X",
"price_bucket": None,
},
"horizon_months": 12,
"base_pace_units_per_mo": 10.0,
"demand_norm_coefficient": 0.8,
"macro_coefficient": 1.1,
"projected_demand_units": 105.6,
"open_units": 300,
"hidden_release_units": 133.33,
"future_online_units": 40.0,
"projected_supply_units": 367.73,
"balance_units": -262.13,
"balance_ratio": 0.287,
"deficit_index": -0.9,
"rate_future": 21.0,
"rate_sensitivity_phrase": "при росте ставки …",
"future_competitors": [{"obj_id": 1, "relevance_weight": 0.7}],
"advisory": True,
"confidence": "medium",
}
base.update(over)
return DemandSupplyForecast(**base) # type: ignore[arg-type]
class TestAsDict:
def test_all_fields_rounded(self) -> None:
d = _make_forecast().as_dict()
assert d["base_pace_units_per_mo"] == 10.0
assert d["demand_norm_coefficient"] == 0.8
assert d["macro_coefficient"] == 1.1
assert d["projected_demand_units"] == 105.6
assert d["projected_supply_units"] == 367.7
assert d["deficit_index"] == -0.9
assert d["advisory"] is True
assert d["confidence"] in _ALLOWED_CONFIDENCE
def test_none_fields_survive(self) -> None:
d = _make_forecast(
base_pace_units_per_mo=None,
projected_demand_units=None,
balance_units=None,
balance_ratio=None,
deficit_index=None,
).as_dict()
assert d["base_pace_units_per_mo"] is None
assert d["projected_demand_units"] is None
assert d["balance_ratio"] is None
assert d["deficit_index"] is None
def test_competitors_passthrough(self) -> None:
d = _make_forecast().as_dict()
assert d["future_competitors"] == [{"obj_id": 1, "relevance_weight": 0.7}]
# ── orchestrator helpers (стабы §9.x compute_*) ───────────────────────────────
def _metrics_stub(*, unit_velocity: float | None = 10.0, confidence: str = "high") -> MagicMock:
m = MagicMock()
m.unit_velocity = unit_velocity
m.confidence = confidence
return m
def _norm_stub(*, coefficient: float = 0.8, confidence: str = "high") -> MagicMock:
m = MagicMock()
m.coefficient = coefficient
m.confidence = confidence
return m
def _macro_coef_stub(*, coefficient: float = 1.1, confidence: str = "high") -> MagicMock:
m = MagicMock()
m.coefficient = coefficient
m.confidence = confidence
return m
def _sens_stub(
*, beta: float = -0.5, x_pct: float = -40.0, phrase: str = "при росте ставки …"
) -> MagicMock:
"""Стаб §9.6: несёт DISTINCTIVE beta/x_pct — они НЕ должны влиять на спрос."""
m = MagicMock()
m.beta = beta
m.x_pct = x_pct
m.phrase = phrase
return m
def _supply_stub(
*,
open_units: int = 300,
hidden_units: int = 200,
future_units_by_horizon: float = 40.0,
confidence: str = "high",
) -> MagicMock:
m = MagicMock()
m.open_units = open_units
m.hidden_units = hidden_units
m.future_units_by_horizon = future_units_by_horizon
m.confidence = confidence
return m
def _competitor_stub(obj_id: int, relevance_weight: float) -> MagicMock:
c = MagicMock()
c.obj_id = obj_id
c.comm_name = f"ЖК-{obj_id}"
c.obj_class = "комфорт"
c.distance_m = 500.0
c.flats_total = 100
c.velocity_per_month = 5.0
c.relevance_weight = relevance_weight
return c
def _competitors_stub(n: int = 3) -> MagicMock:
resp = MagicMock()
resp.competitors = [_competitor_stub(i, 0.9 - 0.1 * i) for i in range(n)]
return resp
class _Patches:
"""Контекст-менеджер: патчит ВСЕ reused §9.x сервисы разом (зеркало духа
test_future_supply @patch, но 7 зависимостей — держим в одном месте).
Стабы передаются по коротким алиасам (_ALIAS) → dotted-path константа цели.
Возвращает dict {dotted_path: MagicMock} для проверки call_args.
"""
# Короткий алиас (kwarg-имя) → реальная dotted-path цель патча.
_ALIAS: ClassVar[dict[str, str]] = {
"_MACRO": _MACRO,
"_METRICS": _METRICS,
"_NORM": _NORM,
"_MACRO_COEF": _MACRO_COEF,
"_SENS": _SENS,
"_SUPPLY": _SUPPLY,
"_COMPETITORS": _COMPETITORS,
}
def __init__(self, **stubs: object) -> None:
self._stubs = stubs
self._ctxs: list[Any] = []
def __enter__(self) -> dict[str, MagicMock]:
defaults: dict[str, object] = {
_MACRO: [_macro_pt(dt.date(2025, 3, 1), 21.0)],
_METRICS: _metrics_stub(),
_NORM: _norm_stub(),
_MACRO_COEF: _macro_coef_stub(),
_SENS: _sens_stub(),
_SUPPLY: _supply_stub(),
_COMPETITORS: _competitors_stub(),
}
for alias, value in self._stubs.items():
target = self._ALIAS.get(alias, alias)
defaults[target] = value
mocks: dict[str, MagicMock] = {}
for target, value in defaults.items():
p = patch(target, return_value=value)
mocks[target] = p.start()
self._ctxs.append(p)
return mocks
def __exit__(self, *exc: object) -> None:
for p in self._ctxs:
p.stop()
def _run(**over: object) -> list[DemandSupplyForecast]:
spec = SegmentSpec(obj_class="комфорт", district="Академический")
return compute_demand_supply_forecast(
MagicMock(),
spec=spec,
district="Академический",
cad_num="66:41:0303161:123",
**over, # type: ignore[arg-type]
)
# ── orchestrator: спрос = pace × §9.4 × §9.5 × h ТОЧНО ─────────────────────────
class TestComputeDemand:
def test_demand_is_pace_times_norm_times_macro_times_h(self) -> None:
with _Patches(
_METRICS=_metrics_stub(unit_velocity=10.0),
_NORM=_norm_stub(coefficient=0.8),
_MACRO_COEF=_macro_coef_stub(coefficient=1.1),
):
res = _run(horizons=[12])
f = res[0]
assert f.base_pace_units_per_mo == 10.0
assert f.demand_norm_coefficient == 0.8
assert f.macro_coefficient == 1.1
# ровно 10 × 0.8 × 1.1 × 12 (ЛИНЕЙНО, без компаунда).
assert f.projected_demand_units == pytest.approx(10.0 * 0.8 * 1.1 * 12)
def test_demand_linear_across_horizons(self) -> None:
with _Patches():
res = _run(horizons=[6, 12, 24])
d6, d12, d24 = (f.projected_demand_units for f in res)
assert d6 is not None and d12 is not None and d24 is not None
# линейность: 12-мес = 2×6-мес, 24-мес = 4×6-мес.
assert d12 == pytest.approx(2.0 * d6)
assert d24 == pytest.approx(4.0 * d6)
def test_none_pace_yields_none_demand(self) -> None:
with _Patches(_METRICS=_metrics_stub(unit_velocity=None)):
res = _run(horizons=[12])
assert res[0].base_pace_units_per_mo is None
assert res[0].projected_demand_units is None
# ── ⚠️ orchestrator: β УЧТЁН РОВНО ОДИН РАЗ (главный тест корректности) ────────
class TestBetaNotDoubleApplied:
"""§9.4 уже применяет β внутри; §9.6 здесь — только для explain-фразы.
projected_demand НЕ должен зависеть от rate_sensitivity.beta/x_pct."""
def _demand_with_sens(self, *, beta: float, x_pct: float) -> float:
with _Patches(
_METRICS=_metrics_stub(unit_velocity=10.0),
_NORM=_norm_stub(coefficient=0.8),
_MACRO_COEF=_macro_coef_stub(coefficient=1.1),
_SENS=_sens_stub(beta=beta, x_pct=x_pct),
):
res = _run(horizons=[12])
demand = res[0].projected_demand_units
assert demand is not None
return demand
def test_demand_independent_of_sensitivity_beta(self) -> None:
# Меняем β/x_pct §9.6 РАДИКАЛЬНО — спрос обязан остаться идентичным
# (β применяется ТОЛЬКО внутри §9.4 demand_norm, не повторно здесь).
mild = self._demand_with_sens(beta=-0.1, x_pct=-5.0)
wild = self._demand_with_sens(beta=-5.0, x_pct=-99.0)
positive = self._demand_with_sens(beta=2.0, x_pct=50.0)
assert mild == pytest.approx(wild)
assert mild == pytest.approx(positive)
def test_demand_equals_pace_norm_macro_h_exactly(self) -> None:
# Спрос = ровно paceק9.4ק9.5×h — НЕТ скрытого ×(1+x_pct) или ×exp(β).
demand = self._demand_with_sens(beta=-3.0, x_pct=-90.0)
assert demand == pytest.approx(10.0 * 0.8 * 1.1 * 12)
def test_sensitivity_phrase_surfaced_for_explain(self) -> None:
# §9.6 ВСЁ ЖЕ используется — но только .phrase в выводе (explainability).
with _Patches(_SENS=_sens_stub(phrase="ставка ↑ → продажи ↓ на 40%")):
res = _run(horizons=[12])
assert res[0].rate_sensitivity_phrase == "ставка ↑ → продажи ↓ на 40%"
def test_norm_coefficient_is_the_only_beta_channel(self) -> None:
# Меняем §9.4 coefficient → спрос ДОЛЖЕН измениться (это единственный β-канал).
with _Patches(_NORM=_norm_stub(coefficient=0.5)):
low = _run(horizons=[12])[0].projected_demand_units
with _Patches(_NORM=_norm_stub(coefficient=1.0)):
high = _run(horizons=[12])[0].projected_demand_units
assert low is not None and high is not None
assert high == pytest.approx(2.0 * low)
# ── orchestrator: предложение (фаза hidden + absorbed-clamp) ───────────────────
class TestComputeSupply:
def test_hidden_release_phased_by_horizon(self) -> None:
# hidden=200; на 6 мес фаза ⅓ → 66.67; future=40; open=300.
with _Patches(
_SUPPLY=_supply_stub(open_units=300, hidden_units=200, future_units_by_horizon=40.0),
_METRICS=_metrics_stub(unit_velocity=0.0), # спрос 0 → absorbed 0 → валовое всё
):
res = _run(horizons=[6])
f = res[0]
assert f.open_units == 300
assert f.hidden_release_units == pytest.approx(200.0 * (1.0 / 3.0))
assert f.future_online_units == pytest.approx(40.0)
# спрос 0 → поглощения нет → чистое = валовое = 300 + 66.67 + 40.
assert f.projected_supply_units == pytest.approx(300.0 + 200.0 / 3.0 + 40.0)
def test_absorbed_reduces_supply_clamped(self) -> None:
# спрос огромный → поглощает всё валовое → чистое предложение 0 (clamp ≥0).
with _Patches(
_METRICS=_metrics_stub(unit_velocity=10_000.0),
_SUPPLY=_supply_stub(open_units=100, hidden_units=100, future_units_by_horizon=50.0),
):
res = _run(horizons=[18])
assert res[0].projected_supply_units == 0.0
def test_supply_never_negative(self) -> None:
with _Patches(_METRICS=_metrics_stub(unit_velocity=500.0)):
for f in _run(horizons=[6, 12, 18, 24]):
assert f.projected_supply_units >= 0.0
# ── orchestrator: баланс / индекс дефицита ─────────────────────────────────────
class TestComputeBalance:
def test_balance_and_ratio_and_index(self) -> None:
# Спрос 10×1.0×1.0×12 = 120; supply: open0+hidden0+future0 → 0 валовое,
# но absorbed=min(0,120)=0 → supply 0 → ratio None, index None.
with _Patches(
_METRICS=_metrics_stub(unit_velocity=10.0),
_NORM=_norm_stub(coefficient=1.0),
_MACRO_COEF=_macro_coef_stub(coefficient=1.0),
_SUPPLY=_supply_stub(open_units=0, hidden_units=0, future_units_by_horizon=0.0),
):
res = _run(horizons=[12])
f = res[0]
assert f.projected_demand_units == pytest.approx(120.0)
assert f.projected_supply_units == 0.0
# supply ≤0 → ratio/index None (НЕ выдуманный ∞-дефицит).
assert f.balance_ratio is None
assert f.deficit_index is None
# balance_units всё равно считается (120 − 0 = 120).
assert f.balance_units == pytest.approx(120.0)
def test_oversupply_negative_index(self) -> None:
# Малый спрос, большое предложение (спрос не поглощает всё) → index < 0.
with _Patches(
_METRICS=_metrics_stub(unit_velocity=1.0),
_NORM=_norm_stub(coefficient=1.0),
_MACRO_COEF=_macro_coef_stub(coefficient=1.0),
_SUPPLY=_supply_stub(open_units=1000, hidden_units=0, future_units_by_horizon=0.0),
):
res = _run(horizons=[12])
f = res[0]
# спрос 12; supply = 1000 − 12 = 988 → ratio 12/988 << 1 → index −1 (затоварка).
assert f.projected_supply_units == pytest.approx(988.0)
assert f.balance_ratio is not None and f.balance_ratio < 1.0
assert f.deficit_index is not None and f.deficit_index < 0.0
# ── orchestrator: confidence = MIN ≤ medium ────────────────────────────────────
class TestComputeConfidence:
def test_all_high_capped_to_medium(self) -> None:
# все компоненты high → MIN high, но advisory-cap роняет до medium.
with _Patches(
_METRICS=_metrics_stub(confidence="high"),
_NORM=_norm_stub(confidence="high"),
_MACRO_COEF=_macro_coef_stub(confidence="high"),
_SUPPLY=_supply_stub(confidence="high"),
):
res = _run(horizons=[12])
assert res[0].confidence == "medium"
def test_low_component_drags_to_low(self) -> None:
# один low (тонкий §9.3) → MIN low (cap не поднимает).
with _Patches(
_METRICS=_metrics_stub(confidence="high"),
_NORM=_norm_stub(confidence="high"),
_MACRO_COEF=_macro_coef_stub(confidence="high"),
_SUPPLY=_supply_stub(confidence="low"),
):
res = _run(horizons=[12])
assert res[0].confidence == "low"
def test_never_exceeds_medium(self) -> None:
for sup_conf in ("high", "medium", "low"):
with _Patches(_SUPPLY=_supply_stub(confidence=sup_conf)):
res = _run(horizons=[12])
assert res[0].confidence in {"low", "medium"}
assert res[0].confidence != "high"
# ── orchestrator: advisory ВСЕГДА True ─────────────────────────────────────────
class TestAdvisoryFlag:
def test_advisory_always_true(self) -> None:
with _Patches():
res = _run(horizons=[6, 12, 18, 24])
assert all(f.advisory is True for f in res)
def test_advisory_true_even_on_thin_data(self) -> None:
with _Patches(
_METRICS=_metrics_stub(unit_velocity=None, confidence="low"),
_SUPPLY=_supply_stub(
open_units=0, hidden_units=0, future_units_by_horizon=0.0, confidence="low"
),
):
res = _run(horizons=[12])
assert res[0].advisory is True
# ── orchestrator: one result per horizon + rate_path ──────────────────────────
class TestHorizonsAndRatePath:
def test_one_result_per_horizon(self) -> None:
with _Patches():
res = _run(horizons=[6, 12, 18, 24])
assert [f.horizon_months for f in res] == [6, 12, 18, 24]
def test_default_horizons(self) -> None:
with _Patches():
res = _run() # без horizons → дефолт (6,12,18,24)
assert [f.horizon_months for f in res] == [6, 12, 18, 24]
def test_explicit_rate_path_used(self) -> None:
# caller-сценарий: разные ставки на разных горизонтах → попадают в rate_future.
with _Patches():
res = _run(horizons=[6, 12], rate_path={6: 15.0, 12: 20.0})
assert res[0].rate_future == 15.0
assert res[1].rate_future == 20.0
def test_default_rate_path_holds_last_rate(self) -> None:
# без rate_path → hold_last_rate из макро-ряда (21.0 в стабе).
with _Patches(_MACRO=[_macro_pt(dt.date(2025, 3, 1), 21.0)]):
res = _run(horizons=[6, 12])
assert res[0].rate_future == 21.0
assert res[1].rate_future == 21.0
def test_norm_called_with_horizon_rate(self) -> None:
# §9.4 получает rate_future именно из rate_path (сценарная ставка).
with _Patches() as mocks:
_run(horizons=[12], rate_path={12: 19.5})
norm_calls = mocks[_NORM].call_args_list
assert any(call.kwargs.get("rate_future") == 19.5 for call in norm_calls)
# ── orchestrator: §9.7 competitors (top-N, graceful) ──────────────────────────
class TestFutureCompetitors:
def test_top_n_competitors_surfaced(self) -> None:
with _Patches(_COMPETITORS=_competitors_stub(n=10)):
res = _run(horizons=[12])
comps = res[0].future_competitors
assert len(comps) == _TOP_COMPETITORS # top-N среза
assert comps[0]["relevance_weight"] == pytest.approx(0.9)
assert "obj_id" in comps[0] and "distance_m" in comps[0]
def test_competitors_request_carries_horizon(self) -> None:
with _Patches() as mocks:
_run(horizons=[24])
# CompetitorsRequest собран с horizon_months=24 (horizon-aware §9.7).
call = mocks[_COMPETITORS].call_args
request = call.args[2]
assert request.horizon_months == 24
def test_competitor_lookup_failure_graceful(self) -> None:
# нет геометрии участка (ValueError) → [] (НЕ валим прогноз).
with _Patches():
with patch(_COMPETITORS, side_effect=ValueError("геометрия не найдена")):
res = _run(horizons=[12])
assert res[0].future_competitors == []
# остальной прогноз всё равно собран.
assert res[0].projected_demand_units is not None
# ── orchestrator: graceful empty → None-поля, никогда не crash ─────────────────
class TestGraceful:
def test_empty_everything_yields_none_fields(self) -> None:
with _Patches(
_MACRO=[],
_METRICS=_metrics_stub(unit_velocity=None, confidence="low"),
_NORM=_norm_stub(coefficient=1.0, confidence="low"),
_MACRO_COEF=_macro_coef_stub(coefficient=1.0, confidence="low"),
_SUPPLY=_supply_stub(
open_units=0, hidden_units=0, future_units_by_horizon=0.0, confidence="low"
),
_COMPETITORS=MagicMock(competitors=[]),
):
res = _run(horizons=[12])
f = res[0]
assert f.base_pace_units_per_mo is None
assert f.projected_demand_units is None
assert f.projected_supply_units == 0.0
assert f.balance_units is None # demand None → balance None
assert f.balance_ratio is None
assert f.deficit_index is None
assert f.future_competitors == []
assert f.confidence == "low"
assert f.advisory is True
def test_rate_future_none_when_no_macro(self) -> None:
# пустой макро-ряд → hold_last_rate даёт None → rate_future None (§9.4 нейтраль).
with _Patches(_MACRO=[]):
res = _run(horizons=[12])
assert res[0].rate_future is None
# demand_norm_coefficient None (rate_future None → §9.4 не применяли как канал).
assert res[0].demand_norm_coefficient is None
def test_returns_list_always(self) -> None:
with _Patches():
res = _run(horizons=[6, 12])
assert isinstance(res, list)
assert all(isinstance(f, DemandSupplyForecast) for f in res)
# ── sanity: math import used (deficit lib) ────────────────────────────────────
def test_math_log_used_in_deficit() -> None:
# _deficit_index использует math.log — sanity на знаковую лог-шкалу.
assert _deficit_index(math.e ** math.log(_DEFICIT_RATIO_SATURATION)) == pytest.approx(1.0)