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

feat(forecasting): §14.2 product scoring card (#985, 954-B) #1017

Merged
bot-backend merged 1 commit from feat/954b-product-scoring into main 2026-06-03 08:11:55 +00:00
Conversation 0 Commits 1 Files changed 3 +1820
3 changed files with 1820 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -19,6 +19,9 @@
нагрузки (субсид. ставка, дохода нет → low-confidence; СБОРКА, ADVISORY).
• scenarios (#984/954-A) — §11 три макро-сценария (conservative/base/
aggressive) прогоном #952 под тремя конвертами ставки (СБОРКА, ADVISORY).
• product_scoring (#985/954-B) — §14.2 десять продуктовых скоров ∈ [0,1] (выше=
лучше) + взвешенный overall (renorm над доступными) + §16 причина на скор;
сводит #950…#984 + live-стек, graceful-None, ADVISORY.
Источники данных:
• макро — таблица macro_indicator через reader site_finder/macro.py (reuse).
@ -59,6 +62,11 @@ from app.services.forecasting.macro_series import (
is_confounded_window,
macro_at_lag,
)
from app.services.forecasting.product_scoring import (
ProductScore,
ProductScoreCard,
compute_score_card,
)
from app.services.forecasting.rate_sensitivity import (
RateSensitivity,
best_lag,
@ -92,6 +100,8 @@ __all__ = [
"MacroCoefficient",
"MonthlyMacro",
"MortgageAffordabilityIndex",
"ProductScore",
"ProductScoreCard",
"RankedSegment",
"RateSensitivity",
"SalesSeries",
@ -109,6 +119,7 @@ __all__ = [
"compute_macro_coefficient",
"compute_rate_sensitivity",
"compute_scenarios",
"compute_score_card",
"f_issuance",
"f_mortgage_rate",
"f_overdue",

899
backend/app/services/forecasting/product_scoring.py Normal file
View file

@ -0,0 +1,899 @@
"""§14.2 продуктовые скоры — детерминированная СОВЕТУЮЩАЯ карта пригодности продукта.
#985 (954-B, Site Finder v2 / «GG-форсайт» ТЗ §14.2), EPIC 10 «Скоринг продукта».
Это **СБОРОЧНЫЙ слой**: своей §9.x-математики НЕ пересобирает — берёт уже-смерженные
сервисы форсайта (#950…#984) и live-стека Site Finder и СВОДИТ их в 10 нормированных
скоров ∈ [0,1] (выше = лучше) + взвешенный overall + §16-подобную короткую причину на
каждый скор. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM; своего SQL НЕТ (reuse сервисов).
10 СКОРОВ (ТЗ §14.2) и их backing-сервис (каждый ∈ [0,1], выше = лучше для девелопера):
1. market_fit ← deficit_index целевого сегмента (#980/#952 спрос↔предложение).
Выше дефицит (недосток) → выше fit. Remap [−1,+1]→[0,1].
2. demand ← market_metrics.unit_velocity (§9.2 абсорбция). Saturating.
3. supply_risk ← future_supply.index + market_metrics.overstock_index.
Выше давление/затоварка → НИЖЕ скор (ИНВЕРТИРОВАН).
4. future_competition ← competitors.relevance_weight density + кол-во конкурентов.
Больше/сильнее будущих конкурентов → НИЖЕ скор (ИНВЕРТИРОВАН).
5. price_feasibility ← affordability.monthly_payment_rub (§7.9 прокси нагрузки).
Меньше платёж → доступнее → выше скор. Saturating-инверс платежа.
6. infra_fit ← poi_score top-7 суммарный weight (Site Finder B6). Выше → выше.
7. mortgage_sensitivity← rate_sensitivity.x_pct (§9.6). Чем отрицательнее (чувствительнее)
→ НИЖЕ скор / флаг риска (ИНВЕРТИРОВАН по магнитуде).
8. differentiation ← USP-белое-пятно из recommendation (§10.5): число позитив-дефицит
USP-сегментов. Больше white-space → выше скор.
9. commercial ← recommendation коммерческий сигнал §10.4 (доля коммерции).
ДЕГРАДИРУЕТ в None, когда сигнал недоступен (НЕ 0-как-заглушка).
10. confidence ← META-скор: MIN/агрегат confidence вкладывающих сервисов +
доля доступных скоров (data-quality). Выше → надёжнее карта.
ИНВЕРТИРОВАННЫЕ (high-bad → low-score): supply_risk, future_competition,
mortgage_sensitivity. Остальные — high-good → high-score напрямую.
РЕНОРМАЛИЗАЦИЯ ВЕСОВ (критично, зеркало §9.5 macro_coefficient.renormalize_contributions):
overall = взвешенное среднее ТОЛЬКО по ДОСТУПНЫМ (value != None) скорам; веса доступных
делятся на свою сумму (renorm) → пропуск скора (тонкие данные) НЕ тянет overall к 0
искусственно. Все скоры None → overall=None честно. confidence-скор в overall НЕ
взвешивается весом-вкладом — он мета-скор о надёжности, а не «качество продукта», но он
ПРИСУТСТВУЕТ в карте как 10-й скор (его вес в _SCORE_WEIGHTS = 0.0, см. ниже).
ADVISORY (КРИТИЧНО — честность, зеркало demand_supply_forecast / recommendation):
карта наследует advisory-статус §9.x (движок не провалидирован до бэктеста #951).
Поэтому `advisory` ВСЕГДА True. Цифры — для explainability/прототипа, НЕ основание для
инвест-решения. Каждый backing-сервис уже advisory-capped; здесь cap не задираем.
Graceful-on-thin-data (дух всех §9.x-сервисов): любой тонкий вход / сбой backing-сервиса
→ соответствующий скор value=None + source='unavailable' (НИКОГДА 0-как-заглушка),
НИКОГДА не валит карту. Детерминированно (без рандома). numpy НЕ нужен.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.schemas.parcel import CompetitorsRequest
from app.services.forecasting.affordability import compute_affordability
from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast
from app.services.forecasting.rate_sensitivity import compute_rate_sensitivity
from app.services.forecasting.recommendation import build_forecast_overlay
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
from app.services.site_finder.poi_score import compute_poi_weighted_top7
if TYPE_CHECKING: # избегаем рантайм-зависимости (типы только для аннотаций)
from app.services.forecasting.demand_supply_forecast import DemandSupplyForecast
logger = logging.getLogger(__name__)
Confidence = Literal["high", "medium", "low"]
# ── Ключи 10 скоров (named — переименование не рассинхронит веса/карту молча) ──
_K_MARKET_FIT: str = "market_fit"
_K_DEMAND: str = "demand"
_K_SUPPLY_RISK: str = "supply_risk"
_K_FUTURE_COMPETITION: str = "future_competition"
_K_PRICE_FEASIBILITY: str = "price_feasibility"
_K_INFRA_FIT: str = "infra_fit"
_K_MORTGAGE_SENSITIVITY: str = "mortgage_sensitivity"
_K_DIFFERENTIATION: str = "differentiation"
_K_COMMERCIAL: str = "commercial"
_K_CONFIDENCE: str = "confidence"
# Канонический порядок 10 скоров (детерминизм карты / тестов).
_SCORE_KEYS: tuple[str, ...] = (
_K_MARKET_FIT,
_K_DEMAND,
_K_SUPPLY_RISK,
_K_FUTURE_COMPETITION,
_K_PRICE_FEASIBILITY,
_K_INFRA_FIT,
_K_MORTGAGE_SENSITIVITY,
_K_DIFFERENTIATION,
_K_COMMERCIAL,
_K_CONFIDENCE,
)
# ── Веса overall (СУММА = 1.0) — экспертная оценка вклада в «пригодность продукта».
# market_fit (баланс спрос↔предложение) — доминирующий сигнал §9.7 «что строить».
# Инвертированные риск-скоры (supply_risk / future_competition / mortgage_sensitivity)
# несут существенную, но меньшую массу. confidence — МЕТА-скор (надёжность, не качество
# продукта) → вес 0.0 в overall, но он ПРИСУТСТВУЕТ как 10-й скор карты. Веса заданы в
# ПОЛНОЙ схеме; renorm делит на сумму ДОСТУПНЫХ (см. _weighted_overall). Эвристика
# (НЕ фит) — уточняется бэктестом #951; карта advisory до валидации. Tunable.
_SCORE_WEIGHTS: dict[str, float] = {
_K_MARKET_FIT: 0.22,
_K_DEMAND: 0.16,
_K_SUPPLY_RISK: 0.13,
_K_FUTURE_COMPETITION: 0.12,
_K_PRICE_FEASIBILITY: 0.11,
_K_INFRA_FIT: 0.09,
_K_MORTGAGE_SENSITIVITY: 0.07,
_K_DIFFERENTIATION: 0.06,
_K_COMMERCIAL: 0.04,
_K_CONFIDENCE: 0.0, # МЕТА-скор: в overall не вкладывается весом (надёжность ≠ качество)
}
# ── Named-нормировочные константы (каждая — задокументированная шкала) ─────────
# market_fit: deficit_index ∈ [−1,+1] (−1 затоварка / +1 сильный дефицит) → [0,1]
# линейным remap (di+1)/2. 0.5 = баланс. Прямая (не насыщающая) — индекс УЖЕ насыщен
# в §9.8 (_deficit_index лог-clamp), повторно не жмём.
_MARKET_FIT_MIDPOINT: float = 0.5 # значение скора при deficit_index = 0 (баланс)
# demand: unit_velocity (ед./мес) → saturating v/(v+half). При v = _DEMAND_HALF_VELOCITY
# скор = 0.5. 8 ед./мес ≈ «здоровый» месячный темп одного ЖК масс-рынка ЕКБ (зеркало
# духа market_metrics абсорбции); ниже → продукт медленнее поглощается → ниже скор.
_DEMAND_HALF_VELOCITY: float = 8.0
# supply_risk (ИНВЕРТИРОВАН): берём СРЕДНЕЕ доступных из {future_supply.index,
# overstock_index} (оба уже ∈ [0,1], выше = хуже) → риск r ∈ [0,1]; скор = 1 − r.
# Оба None → None. Среднее (а не max) — оба канала равноправно описывают избыток
# предложения (грядущее давление + зависший сток). Прямая инверсия (1−r) — каналы
# уже нормированы.
# (без доп. константы — оба входа уже в [0,1])
# future_competition (ИНВЕРТИРОВАН): плотность конкурентов = mean(top-N relevance_weight)
# масштабируем кол-вом найденных через count/(count+half). density ∈ [0,1];
# скор = 1 − density. _COMP_HALF_COUNT = «при стольких релевантных конкурентах
# масштаб-множитель = 0.5». 5 ≈ умеренно плотный рынок в радиусе 1 км. Нет конкурентов
# → density=0 → скор 1.0 (чистое поле — хорошо). Пустой ответ (нет геометрии) → None.
_COMP_HALF_COUNT: float = 5.0
_COMP_TOP_N: int = 5 # сколько верхних по relevance_weight усредняем (зеркало §9.8 top-N)
# price_feasibility: monthly_payment_rub (₽/мес, прокси нагрузки §7.9) → saturating
# ИНВЕРС: скор = half/(half+payment), т.е. меньше платёж → выше скор. При payment =
# _PRICE_HALF_PAYMENT_RUB скор = 0.5. 60 000 ₽/мес ≈ ориентир «комфортного» аннуитета
# масс-рынка ЕКБ на эталонную квартиру (НЕ привязка к доходу — дохода в данных нет,
# #946; это ОТНОСИТЕЛЬНАЯ шкала тяжести платежа). Платёж None → None.
_PRICE_HALF_PAYMENT_RUB: float = 60_000.0
# infra_fit: Σ top-7 POI weight (poi_score формула 1/(d+100)·cat_weight) → saturating
# w/(w+half). При сумме = _INFRA_HALF_WEIGHT скор = 0.5. 0.06 ≈ умеренно обеспеченная
# инфраструктурой локация (несколько близких POI среднего веса); калибровано под
# масштаб poi_score weight (метро на ~400 м даёт ~0.012, школа рядом ~0.01). Нет POI /
# нет координат участка → None (НЕ 0 — «нет данных» ≠ «нет инфраструктуры рядом»).
_INFRA_HALF_WEIGHT: float = 0.06
# mortgage_sensitivity (ИНВЕРТИРОВАН по магнитуде): x_pct = % изменения продаж на +1 п.п.
# ставки (NEGATIVE при чувствительном сегменте, §9.6). Риск = min(1, |x_pct|/max);
# скор = 1 − риск. _MORTG_MAX_X_PCT = «падение продаж на 25% за +1 п.п. — уже полный
# риск-флаг» (экстремально rate-чувствительный сегмент). x_pct None → None. Знак x_pct≥0
# (продажи якобы растут при росте ставки — не доверяем, §9.6) трактуем как нет риска
# (риск 0 → скор 1.0): мы НЕ штрафуем за «нечувствительность».
_MORTG_MAX_X_PCT: float = 25.0
# differentiation: число позитив-дефицит USP-сегментов (§10.5) из overlay →
# n/_DIFF_TARGET_USP, clamp [0,1]. 3 USP-пункта (= _USP_TOP_K в recommendation) =
# полный white-space сигнал (скор 1.0). 0 USP → 0.0 (нет белых пятен — слабая
# дифференциация). Overlay недоступен/сбой → None (НЕ 0 — нет данных ≠ нет ниш).
_DIFF_TARGET_USP: float = 3.0
# commercial: commercial_share_pct ∈ [0,100] (§10.4 реализованная доля нежилого) → /100.
# Сигнал ДЕГРАДИРОВАН по дизайну (objective ~ жильё): overlay.commercial.available=False
# → скор None (НЕ 0). available=True → share/100 как прокси силы спроса на коммерцию.
# confidence (МЕТА-скор): маппинг confidence-меток вкладывающих сервисов в [0,1] →
# усредняем с долей доступных скоров (data-quality). high=1.0 / medium=0.6 / low=0.25 —
# монотонная шкала (зеркало рангов §9.x). Доля доступных = (кол-во value!=None из 9
# продуктовых скоров) / 9. Финал = mean(label-score, availability-fraction).
_CONF_LABEL_VALUE: dict[Confidence, float] = {"high": 1.0, "medium": 0.6, "low": 0.25}
# Радиус поиска конкурентов (км) — дефолт CompetitorsRequest (прямые соседи). Явная
# константа для детерминизма вызова (зеркало demand_supply_forecast._COMPETITORS_RADIUS_KM).
_COMPETITORS_RADIUS_KM: float = 1.0
# Радиус POI (м) для infra_fit — дефолт poi_score (2 км пешей доступности). Явно.
_POI_RADIUS_M: int = 2000
# Горизонт по умолчанию (мес) — зеркало продуктового «на горизонте» (#982/#983 default 12).
_DEFAULT_HORIZON_MONTHS: int = 12
# Порядок уверенности для MIN-агрегации (хуже = ниже). Зеркало §9.x-сервисов.
_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}
_RANK_TO_CONFIDENCE: dict[int, Confidence] = {0: "low", 1: "medium", 2: "high"}
# SQL центроида участка (lat/lon для poi_score) — reuse семантики competitors
# (cad_parcels_geom → fallback cad_quarters_geom). Параметризован, без f-string.
_PARCEL_CENTROID_SQL = text("""
SELECT ST_X(pt) AS lon, ST_Y(pt) AS lat
FROM (
SELECT ST_Centroid(geom) AS pt
FROM cad_parcels_geom
WHERE cad_num = :cad_num AND geom IS NOT NULL
UNION ALL
SELECT ST_Centroid(geom) AS pt
FROM cad_quarters_geom
WHERE cad_number = :quarter AND geom IS NOT NULL
) sub
LIMIT 1
""")
@dataclass(frozen=True)
class ProductScore:
"""Один из 10 §14.2 продуктовых скоров (считается на лету, детерминированно).
`value` ∈ [0,1] (выше = лучше для девелопера) ИЛИ None, когда backing-данные тонкие
(НИКОГДА 0-как-заглушка). `source` = имя backing-сервиса или 'unavailable'.
`reason` — короткая RU §16-подобная причина. Инвертированные скоры (supply_risk /
future_competition / mortgage_sensitivity) уже приведены к «выше = лучше».
"""
key: str # один из _SCORE_KEYS
value: float | None # ∈ [0,1] или None (тонкие данные)
confidence: Confidence
reason: str # короткая RU §16-причина
source: str # backing-сервис или 'unavailable'
def as_dict(self) -> dict[str, Any]:
return {
"key": self.key,
"value": _round_or_none(self.value, 3),
"confidence": self.confidence,
"reason": self.reason,
"source": self.source,
}
@dataclass(frozen=True)
class ProductScoreCard:
"""§14.2 карта из 10 продуктовых скоров для сегмента/горизонта (считается на лету).
Все 10 скоров присутствуют в `scores` (тонкие → value=None + source='unavailable').
`overall` — взвешенное среднее ТОЛЬКО по доступным скорам с РЕНОРМАЛИЗАЦИЕЙ весов
над доступными (зеркало §9.5): пропуск скора НЕ тянет overall к 0; все скоры None →
overall=None. `advisory` ВСЕГДА True (наследует advisory-cap §9.x). `confidence` —
МИН-агрегат вкладывающих + data-quality. Карта — explainability, НЕ инвест-решение.
"""
segment: dict[str, str | None]
horizon_months: int
scores: dict[str, ProductScore] # все 10 (по _SCORE_KEYS)
overall: float | None # взвешенное среднее доступных (renorm над доступными)
advisory: bool # ВСЕГДА True
confidence: Confidence
def as_dict(self) -> dict[str, Any]:
return {
"segment": dict(self.segment),
"horizon_months": self.horizon_months,
"scores": {k: self.scores[k].as_dict() for k in _SCORE_KEYS if k in self.scores},
"overall": _round_or_none(self.overall, 3),
"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
def _clamp01(value: float) -> float:
"""clamp в [0,1]. PURE."""
return max(0.0, min(1.0, value))
def _saturating(value: float | None, half: float) -> float | None:
"""Насыщение неотрицательной величины → [0,1): v/(v+half). PURE.
При value = half → 0.5; монотонно неубывающее; стремится к 1.0 при value→∞.
value None / half ≤ 0 → None (нет осмысленной шкалы — не выдумываем). value < 0
клампим к 0 (величины-входы неотрицательны по смыслу).
"""
if value is None or half <= 0:
return None
v = max(0.0, value)
return v / (v + half)
def _unavailable(key: str, reason: str) -> ProductScore:
"""Скор-заглушка для тонких данных: value=None, source='unavailable', conf='low'. PURE."""
return ProductScore(key=key, value=None, confidence="low", reason=reason, source="unavailable")
# ──────────────────────────────────────────────────────────────────────────────
# Pure-скореры — по одному на скор. Каждый: backing-выход → (value|None, conf, reason).
# None-вход → None + 'unavailable'-причина (НИКОГДА 0). Инверсия где помечено.
# PURE, без БД, полностью юнит-тестируемо.
# ──────────────────────────────────────────────────────────────────────────────
def _score_market_fit(
deficit_index: float | None, confidence: Confidence
) -> tuple[float | None, Confidence, str]:
"""market_fit ← deficit_index ∈ [−1,+1] → [0,1] линейным remap (di+1)/2. PURE.
Выше дефицит (недосток предложения) → выше fit. 0.5 = баланс. deficit_index None
(тонкое/неизмеримое предложение) → None. Прямой (не инвертирован).
"""
if deficit_index is None:
return None, "low", "Дефицит предложения по сегменту неизмерим (тонкие данные §9.8)."
value = _clamp01((deficit_index + 1.0) / 2.0)
if deficit_index > 0:
reason = f"Дефицит предложения (индекс {round(deficit_index, 2)}) — рынок недонасыщен."
elif deficit_index < 0:
reason = f"Затоварка (индекс {round(deficit_index, 2)}) — предложения больше спроса."
else:
reason = "Баланс спроса и предложения по сегменту."
return value, confidence, reason
def _score_demand(
unit_velocity: float | None, confidence: Confidence
) -> tuple[float | None, Confidence, str]:
"""demand ← unit_velocity (ед./мес, §9.2) → saturating v/(v+half). PURE.
Выше темп поглощения → выше скор. None → None (нет выборки ≠ нулевой спрос).
"""
value = _saturating(unit_velocity, _DEMAND_HALF_VELOCITY)
if value is None:
return None, "low", "Темп продаж по локации неизмерим (нет выборки §9.2)."
reason = f"Темп поглощения ~{round(unit_velocity or 0.0, 1)} ед./мес по локации."
return value, confidence, reason
def _score_supply_risk(
future_pressure_index: float | None,
overstock_index: float | None,
confidence: Confidence,
) -> tuple[float | None, Confidence, str]:
"""supply_risk (ИНВЕРТИРОВАН) ← mean(future_supply.index, overstock_index) → 1−риск. PURE.
Оба входа ∈ [0,1], выше = хуже (больше грядущего давления / зависшего стока).
Риск = среднее доступных; скор = 1 − риск (high-bad → low-score). Оба None → None.
"""
parts = [p for p in (future_pressure_index, overstock_index) if p is not None]
if not parts:
return None, "low", "Давление будущего предложения и затоварка неизмеримы (§9.3)."
risk = _clamp01(sum(parts) / len(parts))
value = 1.0 - risk
reason = (
f"Риск избытка предложения (давление+затоварка) ~{round(risk, 2)} — "
f"{'низкий' if risk < 0.5 else 'повышенный'}."
)
return value, confidence, reason
def _score_future_competition(
relevance_weights: list[float] | None,
n_competitors: int | None,
confidence: Confidence,
) -> tuple[float | None, Confidence, str]:
"""future_competition (ИНВЕРТИРОВАН) ← плотность конкурентов → 1−density. PURE.
density = mean(top-N relevance_weight) × count/(count+half). Больше/сильнее будущих
конкурентов → выше density → НИЖЕ скор. relevance_weights None (нет геометрии/сбой)
→ None. Пустой список (нет конкурентов) → density 0 → скор 1.0 (чистое поле).
"""
if relevance_weights is None or n_competitors is None:
return None, "low", "Будущие конкуренты не определены (нет геометрии участка §9.7)."
count = max(0, n_competitors)
if count == 0 or not relevance_weights:
return 1.0, confidence, "Релевантных будущих конкурентов в радиусе не найдено."
top = sorted(relevance_weights, reverse=True)[:_COMP_TOP_N]
mean_weight = _clamp01(sum(top) / len(top))
count_scale = count / (count + _COMP_HALF_COUNT)
density = _clamp01(mean_weight * count_scale)
value = 1.0 - density
reason = (
f"{count} релевантных будущих конкурентов (плотность ~{round(density, 2)}) — "
f"конкуренция {'умеренная' if density < 0.5 else 'высокая'}."
)
return value, confidence, reason
def _score_price_feasibility(
monthly_payment_rub: float | None, confidence: Confidence
) -> tuple[float | None, Confidence, str]:
"""price_feasibility ← monthly_payment_rub (₽/мес, §7.9 прокси) → saturating ИНВЕРС. PURE.
Меньше платёж → доступнее → ВЫШЕ скор: скор = half/(half+payment). None → None.
Шкала ОТНОСИТЕЛЬНАЯ (дохода в данных нет, #946) — тяжесть платежа, не payment/income.
"""
if monthly_payment_rub is None or monthly_payment_rub < 0:
return None, "low", "Платёж по ипотеке неизмерим (нет цены сегмента §7.9)."
value = _PRICE_HALF_PAYMENT_RUB / (_PRICE_HALF_PAYMENT_RUB + monthly_payment_rub)
reason = (
f"Прокси платежа ~{round(monthly_payment_rub / 1000.0)} тыс.₽/мес "
f"(субсид. ставка, §7.9 — доступность относительная)."
)
return value, confidence, reason
def _score_infra_fit(
poi_weight_sum: float | None, confidence: Confidence
) -> tuple[float | None, Confidence, str]:
"""infra_fit ← Σ top-7 POI weight (poi_score B6) → saturating w/(w+half). PURE.
Выше суммарный вес близких POI → выше скор. None (нет координат участка / нет POI)
→ None (нет данных ≠ нет инфраструктуры). Прямой (не инвертирован).
"""
value = _saturating(poi_weight_sum, _INFRA_HALF_WEIGHT)
if value is None:
return None, "low", "Инфраструктура (POI) вокруг участка не определена."
reason = f"Сумма весов близких POI ~{round(poi_weight_sum or 0.0, 3)} (метро/школы/ТЦ B6)."
return value, confidence, reason
def _score_mortgage_sensitivity(
x_pct: float | None, confidence: Confidence
) -> tuple[float | None, Confidence, str]:
"""mortgage_sensitivity (ИНВЕРТИРОВАН по магнитуде) ← x_pct (§9.6) → 1−риск. PURE.
x_pct = % изменения продаж на +1 п.п. ставки (NEGATIVE = чувствителен). Риск =
min(1, |x_pct|/max) ТОЛЬКО для x_pct<0; скор = 1 − риск (чувствительнее → НИЖЕ).
x_pct ≥ 0 (не доверяем «растёт при росте ставки», §9.6) → риск 0 → скор 1.0
(не штрафуем за нечувствительность). None → None.
"""
if x_pct is None:
return None, "low", "Чувствительность к ключевой ставке не определена (§9.6)."
if x_pct >= 0:
return 1.0, confidence, "Сегмент малочувствителен к ключевой ставке (§9.6)."
risk = _clamp01(abs(x_pct) / _MORTG_MAX_X_PCT)
value = 1.0 - risk
reason = (
f"Продажи −{round(abs(x_pct), 1)}% на +1 п.п. ставки (§9.6) — "
f"чувствительность {'умеренная' if risk < 0.5 else 'высокая'}."
)
return value, confidence, reason
def _score_differentiation(
n_positive_usp: int | None, confidence: Confidence
) -> tuple[float | None, Confidence, str]:
"""differentiation ← число позитив-дефицит USP-сегментов (§10.5) → n/target. PURE.
Больше white-space (недообеспеченных ниш) → выше скор. n None (overlay сбой/
недоступен) → None. n=0 → 0.0 (нет белых пятен — слабая дифференциация). Прямой.
"""
if n_positive_usp is None:
return None, "low", "USP-ниши (§10.5) не определены (overlay недоступен)."
value = _clamp01(n_positive_usp / _DIFF_TARGET_USP)
white_space = "есть" if n_positive_usp else "не выражен"
reason = f"{n_positive_usp} дефицит-ниш (USP, §10.5) — white-space {white_space}."
return value, confidence, reason
def _score_commercial(
commercial: dict[str, Any] | None,
) -> tuple[float | None, Confidence, str]:
"""commercial ← recommendation §10.4 сигнал → share_pct/100. ДЕГРАДИРУЕТ в None. PURE.
overlay.commercial.available=True → commercial_share_pct/100 (прокси спроса на
нежилое). available=False / None (objective ~ жильё, тонко) → None (НЕ 0-как-заглушка).
confidence наследуется из сигнала (или 'low').
"""
if not commercial or not commercial.get("available"):
return None, "low", "Коммерческий сигнал §10.4 недоступен (нет данных по нежилому)."
share_pct = commercial.get("commercial_share_pct")
if share_pct is None:
return None, "low", "Коммерческий сигнал §10.4 недоступен (доля не измерена)."
raw_conf = commercial.get("confidence", "low")
conf: Confidence = raw_conf if raw_conf in ("high", "medium", "low") else "low"
value = _clamp01(float(share_pct) / 100.0)
reason = f"Реализованная доля коммерции ~{round(float(share_pct), 1)}% (§10.4, прокси спроса)."
return value, conf, reason
def _score_confidence(
contributing: list[Confidence],
n_available: int,
n_product_scores: int,
) -> tuple[float | None, Confidence, str]:
"""confidence (МЕТА) ← mean(label-score вкладывающих, доля доступных скоров). PURE.
label-score: high=1.0/medium=0.6/low=0.25 (усредняем по вкладывающим сервисам).
availability = n_available / n_product_scores (сколько из 9 продуктовых скоров не
None). Финал value = mean(mean-label, availability). Возвращаемая confidence-метка =
MIN вкладывающих (худший тянет вниз). Пусто/всё-None → ('low'-value по availability).
"""
label_scores = [_CONF_LABEL_VALUE[c] for c in contributing]
mean_label = sum(label_scores) / len(label_scores) if label_scores else 0.0
availability = (n_available / n_product_scores) if n_product_scores > 0 else 0.0
value = _clamp01((mean_label + availability) / 2.0)
label = _min_confidence(contributing)
reason = (
f"Сводная надёжность: {n_available}/{n_product_scores} скоров доступно, "
f"вкладывающие сервисы ~{label}."
)
return value, label, reason
# ──────────────────────────────────────────────────────────────────────────────
# Pure-агрегаты карты — confidence-min, взвешенный overall с renorm над доступными.
# ──────────────────────────────────────────────────────────────────────────────
def _min_confidence(values: list[Confidence]) -> Confidence:
"""Итоговая метка = MIN компонентных (худшая тянет вниз). Пусто → 'low'. PURE."""
ranks = [_CONFIDENCE_RANK[v] for v in values]
if not ranks:
return "low"
return _RANK_TO_CONFIDENCE[min(ranks)]
def _weighted_overall(scores: dict[str, ProductScore], weights: dict[str, float]) -> float | None:
"""Взвешенное среднее ДОСТУПНЫХ скоров с РЕНОРМАЛИЗАЦИЕЙ весов над доступными. PURE.
Зеркало §9.5 renormalize_contributions: доступные (value!=None И вес>0) делят свой
вес на СУММУ доступных весов → пропуск скора НЕ тянет overall к 0 искусственно.
Все скоры None (или нулевые веса) → None честно. Скоры с весом 0 (confidence-мета)
в overall НЕ участвуют. value уже ∈ [0,1], поэтому overall ∈ [0,1]. PURE.
Args:
scores: ключ → ProductScore (value может быть None).
weights: ключ → вес (полная схема; renorm над доступными).
Returns:
Взвешенное среднее ∈ [0,1] или None (нет доступных взвешенных скоров).
"""
available_weight = 0.0
weighted_sum = 0.0
for key, weight in weights.items():
if weight <= 0:
continue
score = scores.get(key)
if score is None or score.value is None:
continue
available_weight += weight
weighted_sum += weight * score.value
if available_weight <= 0:
return None
return _clamp01(weighted_sum / available_weight)
# ──────────────────────────────────────────────────────────────────────────────
# DB-оркестратор — тонкий, graceful. Pure-скореры выше тестируются без него.
# ──────────────────────────────────────────────────────────────────────────────
def compute_score_card(
db: Session,
*,
spec: SegmentSpec,
district: str | None,
cad_num: str,
horizon_months: int = _DEFAULT_HORIZON_MONTHS,
) -> ProductScoreCard:
"""Собрать §14.2 карту из 10 продуктовых скоров для сегмента/участка/горизонта.
ADVISORY (наследует advisory-статус §9.x, см. module docstring) — НЕ для production-
решений. СБОРКА, НЕ пересчёт §9.x: вызывает уже-смерженные сервисы ПО ОДНОМУ разу:
• #980/#952 compute_demand_supply_forecast → deficit_index (market_fit).
• §9.2 compute_market_metrics → unit_velocity (demand) + overstock_index (supply_risk).
• §9.3 compute_future_supply_pressure → index (supply_risk).
• §9.7 get_competitors → relevance_weight density (future_competition).
• §7.9 compute_affordability → monthly_payment_rub (price_feasibility).
• poi_score compute_poi_weighted_top7 → Σ weight (infra_fit; нужен центроид участка).
• §9.6 compute_rate_sensitivity → x_pct (mortgage_sensitivity).
• §10.x build_forecast_overlay → USP §10.5 (differentiation) + commercial §10.4.
Затем кормит pure-скореры, ренормализует веса над доступными → overall, считает
мета-confidence. advisory ВСЕГДА True.
Graceful (КРИТИЧНО): любой backing-сервис тонкий/бросает → СВОЙ скор value=None +
source='unavailable', карта ВСЁ РАВНО возвращается (никогда не crash). Каждый вызов
обёрнут так, что сбой одного сервиса не роняет остальные скоры. Детерминированно.
Args:
db: SQLAlchemy sync Session.
spec: целевой сегмент рынка (любой subset осей).
district: район для §9.2/§9.3 (None → ЕКБ-wide).
cad_num: кадастровый номер участка (вход §9.7 + центроид для POI).
horizon_months: горизонт прогноза (мес; по умолчанию _DEFAULT_HORIZON_MONTHS).
Returns:
ProductScoreCard — всегда 10 скоров (тонкие → None+unavailable), overall
(renorm над доступными), advisory=True, мета-confidence.
"""
segment = spec.as_dict()
scores: dict[str, ProductScore] = {}
# ── market_fit + demand + supply_risk (часть) ← демандо-сапплай + метрики ───
forecast = _safe_call(
"demand_supply_forecast",
lambda: _forecast_at_horizon(
db, spec=spec, district=district, cad_num=cad_num, horizon_months=horizon_months
),
)
metrics = _safe_call(
"market_metrics",
lambda: compute_market_metrics(db, district=district),
)
fsp = _safe_call(
"future_supply",
lambda: compute_future_supply_pressure(
db, district=district, horizon_months=horizon_months
),
)
deficit_index = forecast.deficit_index if forecast is not None else None
market_conf = forecast.confidence if forecast is not None else "low"
scores[_K_MARKET_FIT] = _build(
"market_fit",
_K_MARKET_FIT,
"demand_supply_forecast",
_score_market_fit(deficit_index, market_conf),
)
unit_velocity = metrics.unit_velocity if metrics is not None else None
demand_conf = metrics.confidence if metrics is not None else "low"
scores[_K_DEMAND] = _build(
"demand", _K_DEMAND, "market_metrics", _score_demand(unit_velocity, demand_conf)
)
overstock = metrics.overstock_index if metrics is not None else None
pressure_index = fsp.index if fsp is not None else None
supply_conf = _min_confidence(
[
c
for c in (
fsp.confidence if fsp is not None else None,
metrics.confidence if metrics is not None else None,
)
if c is not None
]
)
scores[_K_SUPPLY_RISK] = _build(
"supply_risk",
_K_SUPPLY_RISK,
"future_supply+market_metrics",
_score_supply_risk(pressure_index, overstock, supply_conf),
)
# ── future_competition ← конкуренты участка (relevance_weight density) ──────
weights, n_comp = _competitor_signal(db, cad_num=cad_num, horizon_months=horizon_months)
scores[_K_FUTURE_COMPETITION] = _build(
"future_competition",
_K_FUTURE_COMPETITION,
"competitors",
_score_future_competition(weights, n_comp, "medium"),
)
# ── price_feasibility ← affordability (§7.9 платёж) ────────────────────────
afford = _safe_call("affordability", lambda: compute_affordability(db, spec=spec))
payment = afford.monthly_payment_rub if afford is not None else None
afford_conf: Confidence = afford.confidence if afford is not None else "low"
scores[_K_PRICE_FEASIBILITY] = _build(
"price_feasibility",
_K_PRICE_FEASIBILITY,
"affordability",
_score_price_feasibility(payment, afford_conf),
)
# ── infra_fit ← poi_score top-7 Σ weight (нужен центроид) ──────────────────
poi_sum = _poi_weight_sum(db, cad_num=cad_num)
scores[_K_INFRA_FIT] = _build(
"infra_fit", _K_INFRA_FIT, "poi_score", _score_infra_fit(poi_sum, "medium")
)
# ── mortgage_sensitivity ← §9.6 x_pct ──────────────────────────────────────
sensitivity = _safe_call("rate_sensitivity", lambda: compute_rate_sensitivity(db, spec=spec))
x_pct = sensitivity.x_pct if sensitivity is not None else None
sens_conf: Confidence = sensitivity.confidence if sensitivity is not None else "low"
scores[_K_MORTGAGE_SENSITIVITY] = _build(
"mortgage_sensitivity",
_K_MORTGAGE_SENSITIVITY,
"rate_sensitivity",
_score_mortgage_sensitivity(x_pct, sens_conf),
)
# ── differentiation + commercial ← overlay (§10.5 USP / §10.4 commercial) ──
overlay = _safe_call(
"recommendation_overlay",
lambda: build_forecast_overlay(
db,
district=district,
cad_num=cad_num,
horizon_months=horizon_months,
target_class=spec.obj_class,
),
)
n_usp = _count_positive_usp(overlay)
overlay_conf = _overlay_confidence(overlay)
scores[_K_DIFFERENTIATION] = _build(
"differentiation",
_K_DIFFERENTIATION,
"recommendation",
_score_differentiation(n_usp, overlay_conf),
)
commercial = overlay.get("commercial") if isinstance(overlay, dict) else None
scores[_K_COMMERCIAL] = _build(
"commercial", _K_COMMERCIAL, "recommendation", _score_commercial(commercial)
)
# ── confidence (МЕТА) ← MIN вкладывающих + data-quality ────────────────────
n_product_scores = len(_SCORE_KEYS) - 1 # все, кроме самого confidence
n_available = sum(1 for k in _SCORE_KEYS if k != _K_CONFIDENCE and scores[k].value is not None)
contributing = [
scores[k].confidence
for k in _SCORE_KEYS
if k != _K_CONFIDENCE and scores[k].value is not None
]
scores[_K_CONFIDENCE] = _build(
"confidence",
_K_CONFIDENCE,
"aggregate",
_score_confidence(contributing, n_available, n_product_scores),
)
# ── overall (renorm над доступными) + карта-confidence ─────────────────────
overall = _weighted_overall(scores, _SCORE_WEIGHTS)
card_confidence = _min_confidence(contributing) if contributing else "low"
logger.info(
"product_scoring: segment=%s h=%d overall=%s available=%d/%d confidence=%s (ADVISORY)",
segment,
horizon_months,
_round_or_none(overall, 3),
n_available,
n_product_scores,
card_confidence,
)
return ProductScoreCard(
segment=segment,
horizon_months=horizon_months,
scores=scores,
overall=overall,
advisory=True,
confidence=card_confidence,
)
# ──────────────────────────────────────────────────────────────────────────────
# Тонкие orchestrator-хелперы (graceful-обёртки над backing-сервисами).
# ──────────────────────────────────────────────────────────────────────────────
def _build(
name: str, key: str, source: str, scored: tuple[float | None, Confidence, str]
) -> ProductScore:
"""Свести вывод pure-скорера в ProductScore. value None → source='unavailable'.
name — человекочитаемое имя скорера (для лога/симметрии), key — канон-ключ карты.
"""
value, confidence, reason = scored
resolved_source = source if value is not None else "unavailable"
return ProductScore(
key=key, value=value, confidence=confidence, reason=reason, source=resolved_source
)
def _safe_call(label: str, fn: Any) -> Any:
"""Вызвать backing-сервис graceful: сбой → None + logger.exception (не crash карты).
Любой backing-сервис может бросить (тонкие данные / нет геометрии / сбой SQL) — мы
логируем и возвращаем None, чтобы соответствующий скор стал None+unavailable, а
остальная карта собралась. Намеренно ловим широкий Exception (изоляция одного
сервиса от карты) с обязательным logger.exception — НЕ молчаливое глотание.
"""
try:
return fn()
except Exception:
logger.exception("product_scoring: backing service %s failed → score unavailable", label)
return None
def _forecast_at_horizon(
db: Session,
*,
spec: SegmentSpec,
district: str | None,
cad_num: str,
horizon_months: int,
) -> DemandSupplyForecast | None:
"""Прогон #980 на ОДНОМ горизонте → DemandSupplyForecast (или None при пустом). Thin."""
forecasts = compute_demand_supply_forecast(
db, spec=spec, district=district, cad_num=cad_num, horizons=[horizon_months]
)
return forecasts[0] if forecasts else None
def _competitor_signal(
db: Session, *, cad_num: str, horizon_months: int
) -> tuple[list[float] | None, int | None]:
"""§9.7 relevance_weight'ы будущих конкурентов + их число. Graceful → (None, None).
Нет геометрии участка / сбой → (None, None) → future_competition становится
None+unavailable. Пустой список конкурентов → ([], 0) → скор 1.0 (чистое поле).
"""
try:
response = get_competitors(
db,
cad_num,
CompetitorsRequest(horizon_months=horizon_months, radius_km=_COMPETITORS_RADIUS_KM),
)
except Exception:
logger.exception(
"product_scoring: competitors lookup failed (cad_num=%s) → future_competition n/a",
cad_num,
)
return None, None
weights = [c.relevance_weight for c in response.competitors if c.relevance_weight is not None]
return weights, len(response.competitors)
def _poi_weight_sum(db: Session, *, cad_num: str) -> float | None:
"""Σ top-7 POI weight вокруг участка (poi_score B6). Graceful → None.
Сначала тянем центроид участка (cad_parcels_geom → fallback cad_quarters_geom), затем
compute_poi_weighted_top7. Нет геометрии / нет POI / сбой → None (infra_fit unavailable).
"""
try:
coords = (
db.execute(
_PARCEL_CENTROID_SQL, {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)}
)
.mappings()
.first()
)
except Exception:
logger.exception(
"product_scoring: centroid lookup failed (cad_num=%s) → infra n/a", cad_num
)
return None
if not coords or coords.get("lat") is None or coords.get("lon") is None:
return None
try:
response = compute_poi_weighted_top7(
db, cad_num, float(coords["lat"]), float(coords["lon"]), radius_m=_POI_RADIUS_M
)
except Exception:
logger.exception("product_scoring: poi_score failed (cad_num=%s) → infra n/a", cad_num)
return None
if not response.top_poi:
return None
return sum(item.weight for item in response.top_poi)
def _quarter_from_cad(cad_num: str) -> str:
"""Кадастровый квартал из номера участка (отрезаем последний сегмент). PURE.
'66:41:0303161:123' → '66:41:0303161'. Уже квартал (3 сегмента) / мусор → как есть
(centroid-SQL тогда просто не найдёт строку → None, graceful).
"""
parts = cad_num.split(":")
if len(parts) >= 4:
return ":".join(parts[:3])
return cad_num
def _count_positive_usp(overlay: Any) -> int | None:
"""Число позитив-дефицит USP-сегментов (§10.5) из overlay. Graceful → None.
overlay None / не dict → None (overlay недоступен). Список usp есть (в т.ч. пустой)
→ считаем элементы с deficit_index > 0. Пустой usp при валидном overlay → 0.
"""
if not isinstance(overlay, dict):
return None
usp = overlay.get("usp")
if usp is None:
return None
count = 0
for item in usp:
di = item.get("deficit_index") if isinstance(item, dict) else None
if di is not None and di > 0:
count += 1
return count
def _overlay_confidence(overlay: Any) -> Confidence:
"""MAX confidence среди ranked_segments overlay (зеркало recommendation._best_confidence).
Overlay недоступен/пуст → 'low'. Сигнал дифференциации не слабее сильнейшей ячейки.
"""
if not isinstance(overlay, dict):
return "low"
ranked = overlay.get("ranked_segments") or []
ranks = [
_CONFIDENCE_RANK[c]
for seg in ranked
if isinstance(seg, dict) and (c := seg.get("confidence")) in _CONFIDENCE_RANK
]
return _RANK_TO_CONFIDENCE[max(ranks)] if ranks else "low"

910
backend/tests/services/forecasting/test_product_scoring.py Normal file
View file

@ -0,0 +1,910 @@
"""Unit-тесты §14.2 продуктового скоринга (#985, 954-B, ADVISORY).
Чистые тесты — БЕЗ живой БД (мокаем все backing-сервисы):
• Каждый pure-скорер (_score_*): value ∈ [0,1]; правильное направление инверсии для
supply_risk / future_competition / mortgage_sensitivity (high-bad → low-score);
None-вход → None + 'unavailable'-причина (НИКОГДА 0-как-заглушка).
• Pure-агрегаты: _weighted_overall — renorm над доступными (пропуск скора НЕ тянет
overall к 0; все скоры None → overall None); _min_confidence.
• compute_score_card через @patch ВСЕХ backing-сервисов: 10 скоров присутствуют,
overall взвешен, advisory ВСЕГДА True, graceful (сбой одного сервиса → его скор
unavailable, карта всё равно возвращается).
Детерминированно, без LLM. Мокаем backing-сервисы + db (нет живой БД).
"""
from __future__ import annotations
import os
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from app.services.forecasting.product_scoring import (
_COMP_TOP_N,
_DEFAULT_HORIZON_MONTHS,
_DEMAND_HALF_VELOCITY,
_DIFF_TARGET_USP,
_INFRA_HALF_WEIGHT,
_MORTG_MAX_X_PCT,
_PRICE_HALF_PAYMENT_RUB,
_SCORE_KEYS,
_SCORE_WEIGHTS,
ProductScore,
ProductScoreCard,
_clamp01,
_count_positive_usp,
_min_confidence,
_overlay_confidence,
_quarter_from_cad,
_saturating,
_score_commercial,
_score_confidence,
_score_demand,
_score_differentiation,
_score_future_competition,
_score_infra_fit,
_score_market_fit,
_score_mortgage_sensitivity,
_score_price_feasibility,
_score_supply_risk,
_weighted_overall,
compute_score_card,
)
from app.services.forecasting.sales_series import SegmentSpec
# Пути патча (backing-сервисы импортированы в namespace product_scoring).
_MOD = "app.services.forecasting.product_scoring"
_P_FORECAST = f"{_MOD}.compute_demand_supply_forecast"
_P_METRICS = f"{_MOD}.compute_market_metrics"
_P_FSP = f"{_MOD}.compute_future_supply_pressure"
_P_COMPETITORS = f"{_MOD}.get_competitors"
_P_AFFORD = f"{_MOD}.compute_affordability"
_P_POI = f"{_MOD}.compute_poi_weighted_top7"
_P_SENS = f"{_MOD}.compute_rate_sensitivity"
_P_OVERLAY = f"{_MOD}.build_forecast_overlay"
# ──────────────────────────────────────────────────────────────────────────────
# Pure-хелперы: _clamp01 / _saturating
# ──────────────────────────────────────────────────────────────────────────────
class TestClamp01:
def test_clamps_below(self) -> None:
assert _clamp01(-0.3) == 0.0
def test_clamps_above(self) -> None:
assert _clamp01(1.7) == 1.0
def test_passthrough_in_range(self) -> None:
assert _clamp01(0.42) == 0.42
class TestSaturating:
def test_half_at_half_constant(self) -> None:
assert _saturating(8.0, 8.0) == pytest.approx(0.5)
def test_monotone_increasing(self) -> None:
assert _saturating(4.0, 8.0) < _saturating(16.0, 8.0)
def test_in_unit_range(self) -> None:
for v in (0.0, 1.0, 100.0, 5000.0):
out = _saturating(v, 8.0)
assert out is not None
assert 0.0 <= out < 1.0
def test_none_input_none(self) -> None:
assert _saturating(None, 8.0) is None
def test_nonpositive_half_none(self) -> None:
assert _saturating(5.0, 0.0) is None
def test_negative_value_clamped_to_zero(self) -> None:
assert _saturating(-5.0, 8.0) == 0.0
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 1: market_fit (deficit_index → [0,1] прямой remap)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreMarketFit:
def test_strong_deficit_high(self) -> None:
value, _conf, _r = _score_market_fit(1.0, "medium")
assert value == pytest.approx(1.0)
def test_strong_overstock_low(self) -> None:
value, _conf, _r = _score_market_fit(-1.0, "medium")
assert value == pytest.approx(0.0)
def test_balance_midpoint(self) -> None:
value, _conf, _r = _score_market_fit(0.0, "high")
assert value == pytest.approx(0.5)
def test_higher_deficit_higher_score(self) -> None:
low, _c1, _r1 = _score_market_fit(-0.4, "medium")
high, _c2, _r2 = _score_market_fit(0.6, "medium")
assert low is not None and high is not None
assert high > low
def test_in_range(self) -> None:
for di in (-1.0, -0.5, 0.0, 0.3, 1.0):
value, _c, _r = _score_market_fit(di, "medium")
assert value is not None
assert 0.0 <= value <= 1.0
def test_confidence_passthrough(self) -> None:
_v, conf, _r = _score_market_fit(0.5, "high")
assert conf == "high"
def test_none_input_unavailable(self) -> None:
value, conf, reason = _score_market_fit(None, "medium")
assert value is None
assert conf == "low"
assert "неизмерим" in reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 2: demand (unit_velocity → saturating)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreDemand:
def test_half_at_half_velocity(self) -> None:
value, _c, _r = _score_demand(_DEMAND_HALF_VELOCITY, "medium")
assert value == pytest.approx(0.5)
def test_higher_velocity_higher_score(self) -> None:
low, _c1, _r1 = _score_demand(2.0, "medium")
high, _c2, _r2 = _score_demand(40.0, "medium")
assert low is not None and high is not None
assert high > low
def test_in_range(self) -> None:
for v in (0.0, 1.0, 8.0, 50.0):
value, _c, _r = _score_demand(v, "medium")
assert value is not None
assert 0.0 <= value < 1.0
def test_none_input_unavailable(self) -> None:
value, conf, reason = _score_demand(None, "medium")
assert value is None
assert conf == "low"
assert reason # непустая причина
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 3: supply_risk (ИНВЕРТИРОВАН — high pressure/overstock → low score)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreSupplyRisk:
def test_inverted_high_pressure_low_score(self) -> None:
low_risk, _c1, _r1 = _score_supply_risk(0.1, 0.1, "medium")
high_risk, _c2, _r2 = _score_supply_risk(0.9, 0.9, "medium")
assert low_risk is not None and high_risk is not None
# Инверсия: больше давление/затоварка → НИЖЕ скор.
assert high_risk < low_risk
def test_full_pressure_zero_score(self) -> None:
value, _c, _r = _score_supply_risk(1.0, 1.0, "medium")
assert value == pytest.approx(0.0)
def test_no_pressure_full_score(self) -> None:
value, _c, _r = _score_supply_risk(0.0, 0.0, "medium")
assert value == pytest.approx(1.0)
def test_mean_of_available(self) -> None:
# risk = mean(0.2, 0.8) = 0.5 → score = 0.5.
value, _c, _r = _score_supply_risk(0.2, 0.8, "medium")
assert value == pytest.approx(0.5)
def test_one_channel_only(self) -> None:
# Только future_pressure доступен (overstock None) → risk=0.4 → score=0.6.
value, _c, _r = _score_supply_risk(0.4, None, "medium")
assert value == pytest.approx(0.6)
def test_in_range(self) -> None:
for p, o in ((0.0, 0.0), (0.5, 0.3), (1.0, 1.0), (0.7, None), (None, 0.2)):
value, _c, _r = _score_supply_risk(p, o, "medium")
assert value is not None
assert 0.0 <= value <= 1.0
def test_both_none_unavailable(self) -> None:
value, conf, reason = _score_supply_risk(None, None, "medium")
assert value is None
assert conf == "low"
assert reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 4: future_competition (ИНВЕРТИРОВАН — more/stronger competitors → low score)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreFutureCompetition:
def test_inverted_more_competition_lower_score(self) -> None:
weak, _c1, _r1 = _score_future_competition([0.2, 0.1], 2, "medium")
strong, _c2, _r2 = _score_future_competition([0.9, 0.9, 0.9, 0.9, 0.9], 20, "medium")
assert weak is not None and strong is not None
# Инверсия: больше/сильнее конкурентов → НИЖЕ скор.
assert strong < weak
def test_no_competitors_full_score(self) -> None:
value, _c, _r = _score_future_competition([], 0, "medium")
assert value == pytest.approx(1.0)
def test_in_range(self) -> None:
cases = [([0.5], 1), ([0.9] * 5, 30), ([0.1, 0.2, 0.3], 3)]
for weights, n in cases:
value, _c, _r = _score_future_competition(weights, n, "medium")
assert value is not None
assert 0.0 <= value <= 1.0
def test_only_top_n_averaged(self) -> None:
# 7 весов, но усредняются только top _COMP_TOP_N — низкие хвостовые игнор.
weights = [0.9] * _COMP_TOP_N + [0.0, 0.0]
value, _c, _r = _score_future_competition(weights, len(weights), "medium")
# mean top-N = 0.9; count_scale = 7/12 → density≈0.525 → score≈0.475.
assert value is not None
assert value < 0.5
def test_none_weights_unavailable(self) -> None:
value, conf, reason = _score_future_competition(None, None, "medium")
assert value is None
assert conf == "low"
assert reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 5: price_feasibility (меньше платёж → выше скор; saturating-инверс)
# ──────────────────────────────────────────────────────────────────────────────
class TestScorePriceFeasibility:
def test_half_at_half_payment(self) -> None:
value, _c, _r = _score_price_feasibility(_PRICE_HALF_PAYMENT_RUB, "low")
assert value == pytest.approx(0.5)
def test_lower_payment_higher_score(self) -> None:
cheap, _c1, _r1 = _score_price_feasibility(20_000.0, "low")
pricey, _c2, _r2 = _score_price_feasibility(200_000.0, "low")
assert cheap is not None and pricey is not None
assert cheap > pricey
def test_in_range(self) -> None:
for p in (0.0, 30_000.0, 60_000.0, 500_000.0):
value, _c, _r = _score_price_feasibility(p, "low")
assert value is not None
assert 0.0 < value <= 1.0
def test_none_input_unavailable(self) -> None:
value, conf, reason = _score_price_feasibility(None, "low")
assert value is None
assert conf == "low"
assert reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 6: infra_fit (Σ POI weight → saturating, прямой)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreInfraFit:
def test_half_at_half_weight(self) -> None:
value, _c, _r = _score_infra_fit(_INFRA_HALF_WEIGHT, "medium")
assert value == pytest.approx(0.5)
def test_higher_weight_higher_score(self) -> None:
low, _c1, _r1 = _score_infra_fit(0.01, "medium")
high, _c2, _r2 = _score_infra_fit(0.5, "medium")
assert low is not None and high is not None
assert high > low
def test_in_range(self) -> None:
for w in (0.0, 0.02, 0.06, 1.0):
value, _c, _r = _score_infra_fit(w, "medium")
assert value is not None
assert 0.0 <= value < 1.0
def test_none_input_unavailable(self) -> None:
value, conf, reason = _score_infra_fit(None, "medium")
assert value is None
assert conf == "low"
assert reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 7: mortgage_sensitivity (ИНВЕРТИРОВАН по магнитуде x_pct)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreMortgageSensitivity:
def test_inverted_more_negative_lower_score(self) -> None:
mild, _c1, _r1 = _score_mortgage_sensitivity(-2.0, "medium")
severe, _c2, _r2 = _score_mortgage_sensitivity(-20.0, "medium")
assert mild is not None and severe is not None
# Инверсия по магнитуде: чувствительнее (отрицательнее) → НИЖЕ скор.
assert severe < mild
def test_full_sensitivity_zero_score(self) -> None:
value, _c, _r = _score_mortgage_sensitivity(-_MORTG_MAX_X_PCT, "medium")
assert value == pytest.approx(0.0)
def test_beyond_max_clamped_zero(self) -> None:
value, _c, _r = _score_mortgage_sensitivity(-100.0, "medium")
assert value == pytest.approx(0.0)
def test_non_negative_x_pct_full_score(self) -> None:
# x_pct ≥ 0 (не доверяем «растёт при росте ставки») → нет риска → score 1.0.
value, _c, reason = _score_mortgage_sensitivity(3.0, "medium")
assert value == pytest.approx(1.0)
assert "малочувствителен" in reason
def test_in_range(self) -> None:
for x in (-25.0, -10.0, -1.0, 0.0, 5.0):
value, _c, _r = _score_mortgage_sensitivity(x, "medium")
assert value is not None
assert 0.0 <= value <= 1.0
def test_none_input_unavailable(self) -> None:
value, conf, reason = _score_mortgage_sensitivity(None, "medium")
assert value is None
assert conf == "low"
assert reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 8: differentiation (число USP-ниш → n/target, прямой)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreDifferentiation:
def test_target_usp_full_score(self) -> None:
value, _c, _r = _score_differentiation(int(_DIFF_TARGET_USP), "medium")
assert value == pytest.approx(1.0)
def test_zero_usp_zero_score(self) -> None:
value, _c, _r = _score_differentiation(0, "medium")
assert value == pytest.approx(0.0)
def test_more_usp_higher_score(self) -> None:
low, _c1, _r1 = _score_differentiation(1, "medium")
high, _c2, _r2 = _score_differentiation(3, "medium")
assert low is not None and high is not None
assert high > low
def test_above_target_clamped(self) -> None:
value, _c, _r = _score_differentiation(10, "medium")
assert value == pytest.approx(1.0)
def test_in_range(self) -> None:
for n in (0, 1, 2, 3, 5):
value, _c, _r = _score_differentiation(n, "medium")
assert value is not None
assert 0.0 <= value <= 1.0
def test_none_input_unavailable(self) -> None:
value, conf, reason = _score_differentiation(None, "medium")
assert value is None
assert conf == "low"
assert reason
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 9: commercial (ДЕГРАДИРУЕТ в None, НЕ 0)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreCommercial:
def test_available_share_scaled(self) -> None:
signal = {"available": True, "commercial_share_pct": 40.0, "confidence": "medium"}
value, conf, _r = _score_commercial(signal)
assert value == pytest.approx(0.4)
assert conf == "medium"
def test_in_range(self) -> None:
for share in (0.0, 25.0, 100.0):
value, _c, _r = _score_commercial(
{"available": True, "commercial_share_pct": share, "confidence": "low"}
)
assert value is not None
assert 0.0 <= value <= 1.0
def test_unavailable_signal_degrades_to_none(self) -> None:
# Ключевое отличие commercial: degraded → None, НЕ 0-как-заглушка.
value, conf, reason = _score_commercial({"available": False, "caveat": "x"})
assert value is None
assert conf == "low"
assert reason
def test_none_signal_degrades_to_none(self) -> None:
value, _conf, _r = _score_commercial(None)
assert value is None
def test_missing_share_degrades_to_none(self) -> None:
value, _conf, _r = _score_commercial({"available": True})
assert value is None
def test_bad_confidence_falls_back_low(self) -> None:
value, conf, _r = _score_commercial(
{"available": True, "commercial_share_pct": 10.0, "confidence": "garbage"}
)
assert value is not None
assert conf == "low"
# ──────────────────────────────────────────────────────────────────────────────
# Скорер 10: confidence (META — mean label + availability)
# ──────────────────────────────────────────────────────────────────────────────
class TestScoreConfidence:
def test_all_high_full_availability(self) -> None:
value, label, _r = _score_confidence(["high"] * 9, 9, 9)
# mean_label=1.0, availability=1.0 → value=1.0; label=min=high.
assert value == pytest.approx(1.0)
assert label == "high"
def test_low_label_drags_label(self) -> None:
_v, label, _r = _score_confidence(["high", "medium", "low"], 3, 9)
# MIN-агрегат → 'low'.
assert label == "low"
def test_partial_availability_lowers_value(self) -> None:
full, _l1, _r1 = _score_confidence(["high"] * 9, 9, 9)
partial, _l2, _r2 = _score_confidence(["high"] * 3, 3, 9)
assert full is not None and partial is not None
assert partial < full
def test_in_range(self) -> None:
value, _l, _r = _score_confidence(["medium", "low"], 5, 9)
assert value is not None
assert 0.0 <= value <= 1.0
def test_empty_contributing(self) -> None:
value, label, _r = _score_confidence([], 0, 9)
assert value == pytest.approx(0.0)
assert label == "low"
# ──────────────────────────────────────────────────────────────────────────────
# Pure-агрегат: _min_confidence
# ──────────────────────────────────────────────────────────────────────────────
class TestMinConfidence:
def test_worst_wins(self) -> None:
assert _min_confidence(["high", "medium", "low"]) == "low"
def test_all_high(self) -> None:
assert _min_confidence(["high", "high"]) == "high"
def test_empty_low(self) -> None:
assert _min_confidence([]) == "low"
# ──────────────────────────────────────────────────────────────────────────────
# Pure-агрегат: _weighted_overall — renorm над доступными
# ──────────────────────────────────────────────────────────────────────────────
def _score(key: str, value: float | None) -> ProductScore:
return ProductScore(key=key, value=value, confidence="medium", reason="x", source="svc")
class TestWeightedOverall:
def test_all_available_weighted_mean(self) -> None:
# 2 равновесных скора 0.0 и 1.0 → среднее 0.5.
scores = {"a": _score("a", 0.0), "b": _score("b", 1.0)}
weights = {"a": 0.5, "b": 0.5}
assert _weighted_overall(scores, weights) == pytest.approx(0.5)
def test_missing_score_renormalized_not_dragged(self) -> None:
# b отсутствует (value None): overall = только a (renorm) = 0.8, НЕ занижен.
scores = {"a": _score("a", 0.8), "b": _score("b", None)}
weights = {"a": 0.5, "b": 0.5}
out = _weighted_overall(scores, weights)
assert out == pytest.approx(0.8)
def test_renorm_weights_proportional(self) -> None:
# a(w=0.2,val=1.0) + c(w=0.6,val=0.0); b(w=0.2) отсутствует.
# renorm над {a,c}: (0.2*1 + 0.6*0)/(0.2+0.6) = 0.25.
scores = {
"a": _score("a", 1.0),
"b": _score("b", None),
"c": _score("c", 0.0),
}
weights = {"a": 0.2, "b": 0.2, "c": 0.6}
assert _weighted_overall(scores, weights) == pytest.approx(0.25)
def test_zero_weight_score_excluded(self) -> None:
# confidence-мета (вес 0) НЕ участвует в overall, даже если value задан.
scores = {"a": _score("a", 1.0), "meta": _score("meta", 0.0)}
weights = {"a": 0.5, "meta": 0.0}
assert _weighted_overall(scores, weights) == pytest.approx(1.0)
def test_all_unavailable_none(self) -> None:
scores = {"a": _score("a", None), "b": _score("b", None)}
weights = {"a": 0.5, "b": 0.5}
assert _weighted_overall(scores, weights) is None
def test_in_range(self) -> None:
scores = {"a": _score("a", 0.3), "b": _score("b", 0.9)}
weights = {"a": 0.4, "b": 0.6}
out = _weighted_overall(scores, weights)
assert out is not None
assert 0.0 <= out <= 1.0
class TestScoreWeightsContract:
def test_weights_sum_to_one(self) -> None:
assert sum(_SCORE_WEIGHTS.values()) == pytest.approx(1.0)
def test_weights_cover_all_keys(self) -> None:
assert set(_SCORE_WEIGHTS) == set(_SCORE_KEYS)
def test_confidence_meta_zero_weight(self) -> None:
assert _SCORE_WEIGHTS["confidence"] == 0.0
# ──────────────────────────────────────────────────────────────────────────────
# Pure-хелперы оркестратора: _quarter_from_cad / _count_positive_usp / _overlay_conf
# ──────────────────────────────────────────────────────────────────────────────
class TestQuarterFromCad:
def test_strips_last_segment(self) -> None:
assert _quarter_from_cad("66:41:0303161:123") == "66:41:0303161"
def test_already_quarter_passthrough(self) -> None:
assert _quarter_from_cad("66:41:0303161") == "66:41:0303161"
class TestCountPositiveUsp:
def test_counts_positive_deficit(self) -> None:
overlay = {
"usp": [
{"deficit_index": 0.5},
{"deficit_index": 0.1},
{"deficit_index": -0.2},
{"deficit_index": None},
]
}
assert _count_positive_usp(overlay) == 2
def test_empty_usp_zero(self) -> None:
assert _count_positive_usp({"usp": []}) == 0
def test_none_overlay_none(self) -> None:
assert _count_positive_usp(None) is None
def test_missing_usp_key_none(self) -> None:
assert _count_positive_usp({"ranked_segments": []}) is None
class TestOverlayConfidence:
def test_max_of_ranked(self) -> None:
overlay = {
"ranked_segments": [
{"confidence": "low"},
{"confidence": "medium"},
]
}
assert _overlay_confidence(overlay) == "medium"
def test_empty_low(self) -> None:
assert _overlay_confidence({"ranked_segments": []}) == "low"
def test_none_low(self) -> None:
assert _overlay_confidence(None) == "low"
# ──────────────────────────────────────────────────────────────────────────────
# Orchestrator: compute_score_card via @patch всех backing-сервисов
# ──────────────────────────────────────────────────────────────────────────────
def _forecast_stub(deficit_index: float | None, confidence: str = "medium") -> MagicMock:
f = MagicMock()
f.deficit_index = deficit_index
f.confidence = confidence
return f
def _metrics_stub(
unit_velocity: float | None = 10.0,
overstock_index: float | None = 0.3,
confidence: str = "medium",
) -> MagicMock:
m = MagicMock()
m.unit_velocity = unit_velocity
m.overstock_index = overstock_index
m.confidence = confidence
return m
def _fsp_stub(index: float | None = 0.4, confidence: str = "medium") -> MagicMock:
f = MagicMock()
f.index = index
f.confidence = confidence
return f
def _competitor_stub(relevance_weight: float | None) -> MagicMock:
c = MagicMock()
c.relevance_weight = relevance_weight
return c
def _competitors_response_stub(weights: list[float | None]) -> MagicMock:
r = MagicMock()
r.competitors = [_competitor_stub(w) for w in weights]
return r
def _afford_stub(payment: float | None = 50_000.0, confidence: str = "low") -> MagicMock:
a = MagicMock()
a.monthly_payment_rub = payment
a.confidence = confidence
return a
def _poi_response_stub(weights: list[float]) -> MagicMock:
r = MagicMock()
r.top_poi = [MagicMock(weight=w) for w in weights]
return r
def _sens_stub(x_pct: float | None = -5.0, confidence: str = "medium") -> MagicMock:
s = MagicMock()
s.x_pct = x_pct
s.confidence = confidence
return s
def _overlay_stub(
n_positive_usp: int = 2,
commercial: dict[str, Any] | None = None,
ranked_confidence: str = "medium",
) -> dict[str, Any]:
usp = [{"deficit_index": 0.5} for _ in range(n_positive_usp)]
return {
"horizon_months": 12,
"mode": "demand_supply",
"advisory": True,
"ranked_segments": [{"confidence": ranked_confidence}],
"warnings": [],
"usp": usp,
"commercial": commercial,
}
def _db_with_centroid(lat: float | None = 56.8, lon: float | None = 60.6) -> MagicMock:
"""MagicMock db, чей execute(...).mappings().first() отдаёт центроид участка."""
db = MagicMock()
row = None if lat is None or lon is None else {"lat": lat, "lon": lon}
db.execute.return_value.mappings.return_value.first.return_value = row
return db
def _patch_all(
*,
forecast: Any = None,
metrics: Any = None,
fsp: Any = None,
competitors: Any = None,
afford: Any = None,
poi: Any = None,
sens: Any = None,
overlay: Any = None,
) -> Any:
"""Контекст-менеджер: патчит ВСЕ 8 backing-сервисов их return_value (или дефолтом)."""
commercial = {"available": True, "commercial_share_pct": 30.0, "confidence": "medium"}
forecast_rv = [forecast if forecast is not None else _forecast_stub(0.5)]
competitors_rv = (
competitors if competitors is not None else _competitors_response_stub([0.4, 0.2])
)
overlay_rv = overlay if overlay is not None else _overlay_stub(commercial=commercial)
patchers = [
patch(_P_FORECAST, return_value=forecast_rv),
patch(_P_METRICS, return_value=metrics if metrics is not None else _metrics_stub()),
patch(_P_FSP, return_value=fsp if fsp is not None else _fsp_stub()),
patch(_P_COMPETITORS, return_value=competitors_rv),
patch(_P_AFFORD, return_value=afford if afford is not None else _afford_stub()),
patch(_P_POI, return_value=poi if poi is not None else _poi_response_stub([0.03, 0.02])),
patch(_P_SENS, return_value=sens if sens is not None else _sens_stub()),
patch(_P_OVERLAY, return_value=overlay_rv),
]
return _MultiPatch(patchers)
class _MultiPatch:
def __init__(self, patchers: list[Any]) -> None:
self._patchers = patchers
def __enter__(self) -> None:
for p in self._patchers:
p.start()
def __exit__(self, *exc: Any) -> None:
for p in reversed(self._patchers):
p.stop()
_SPEC = SegmentSpec(obj_class="комфорт", room_bucket="2-к 45-60")
_CAD = "66:41:0303161:123"
class TestComputeScoreCardHappy:
def test_returns_score_card(self) -> None:
with _patch_all():
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="Академический", cad_num=_CAD
)
assert isinstance(card, ProductScoreCard)
def test_all_ten_scores_present(self) -> None:
with _patch_all():
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
assert set(card.scores) == set(_SCORE_KEYS)
assert len(card.scores) == 10
def test_all_values_in_range_when_backed(self) -> None:
with _patch_all():
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
for key in _SCORE_KEYS:
value = card.scores[key].value
assert value is not None, f"{key} unexpectedly unavailable"
assert 0.0 <= value <= 1.0
def test_overall_weighted_in_range(self) -> None:
with _patch_all():
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
assert card.overall is not None
assert 0.0 <= card.overall <= 1.0
def test_advisory_always_true(self) -> None:
with _patch_all():
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
assert card.advisory is True
def test_segment_and_horizon_carried(self) -> None:
with _patch_all():
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD, horizon_months=18
)
assert card.segment == _SPEC.as_dict()
assert card.horizon_months == 18
def test_default_horizon(self) -> None:
with _patch_all():
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
assert card.horizon_months == _DEFAULT_HORIZON_MONTHS
def test_as_dict_shape(self) -> None:
with _patch_all():
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
d = card.as_dict()
assert d["advisory"] is True
assert set(d["scores"]) == set(_SCORE_KEYS)
assert "overall" in d
assert d["scores"]["market_fit"]["source"] == "demand_supply_forecast"
class TestComputeScoreCardGraceful:
def test_forecast_raises_market_fit_unavailable_card_returns(self) -> None:
# Сбой одного сервиса (#980) → market_fit unavailable, КАРТА всё равно собрана.
with _patch_all():
with patch(_P_FORECAST, side_effect=RuntimeError("boom")):
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
assert card.scores["market_fit"].value is None
assert card.scores["market_fit"].source == "unavailable"
# Остальные скоры по-прежнему backed.
assert card.scores["demand"].value is not None
assert card.overall is not None
def test_metrics_raises_demand_and_supply_degrade(self) -> None:
with _patch_all():
with patch(_P_METRICS, side_effect=RuntimeError("boom")):
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
assert card.scores["demand"].value is None
assert card.scores["demand"].source == "unavailable"
# supply_risk теряет overstock-канал, но future_supply.index ещё есть → backed.
assert card.scores["supply_risk"].value is not None
def test_competitors_raises_future_competition_unavailable(self) -> None:
with _patch_all():
with patch(_P_COMPETITORS, side_effect=RuntimeError("no geom")):
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
assert card.scores["future_competition"].value is None
assert card.scores["future_competition"].source == "unavailable"
def test_poi_raises_infra_unavailable(self) -> None:
with _patch_all():
with patch(_P_POI, side_effect=RuntimeError("poi boom")):
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
assert card.scores["infra_fit"].value is None
assert card.scores["infra_fit"].source == "unavailable"
def test_no_centroid_infra_unavailable(self) -> None:
# db без геометрии участка (центроид None) → infra_fit unavailable.
with _patch_all():
card = compute_score_card(
_db_with_centroid(lat=None, lon=None), spec=_SPEC, district="X", cad_num=_CAD
)
assert card.scores["infra_fit"].value is None
assert card.scores["infra_fit"].source == "unavailable"
def test_overlay_raises_differentiation_and_commercial_unavailable(self) -> None:
with _patch_all():
with patch(_P_OVERLAY, side_effect=RuntimeError("overlay boom")):
card = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
assert card.scores["differentiation"].value is None
assert card.scores["commercial"].value is None
assert card.scores["commercial"].source == "unavailable"
def test_commercial_degraded_signal_none_not_zero(self) -> None:
# overlay есть, но commercial.available=False → commercial None (НЕ 0).
degraded = _overlay_stub(commercial={"available": False, "caveat": "thin"})
with _patch_all(overlay=degraded):
card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
assert card.scores["commercial"].value is None
assert card.scores["commercial"].source == "unavailable"
def test_all_services_raise_card_still_returns_overall_none(self) -> None:
# Тотальный сбой: каждый backing бросает → все продуктовые скоры unavailable,
# overall None, но КАРТА возвращается (никогда не crash) с advisory True.
boom = RuntimeError("boom")
db = MagicMock()
db.execute.side_effect = boom
with (
patch(_P_FORECAST, side_effect=boom),
patch(_P_METRICS, side_effect=boom),
patch(_P_FSP, side_effect=boom),
patch(_P_COMPETITORS, side_effect=boom),
patch(_P_AFFORD, side_effect=boom),
patch(_P_POI, side_effect=boom),
patch(_P_SENS, side_effect=boom),
patch(_P_OVERLAY, side_effect=boom),
):
card = compute_score_card(db, spec=_SPEC, district="X", cad_num=_CAD)
assert isinstance(card, ProductScoreCard)
assert card.advisory is True
# future_competition None (signal failed); все 9 продуктовых unavailable.
product_keys = [k for k in _SCORE_KEYS if k != "confidence"]
assert all(card.scores[k].value is None for k in product_keys)
assert card.overall is None
assert card.confidence == "low"
# confidence-мета всё равно посчитан (value не None — это data-quality мера).
assert card.scores["confidence"].value is not None
def test_missing_score_does_not_drag_overall(self) -> None:
# Один сервис падает (market_fit, самый тяжёлый вес) → overall пересчитан renorm
# над оставшимися, НЕ занижен нулём за выпавший market_fit.
with _patch_all():
with patch(_P_FORECAST, side_effect=RuntimeError("boom")):
card_missing = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
card_full = compute_score_card(
_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
)
assert card_missing.overall is not None and card_full.overall is not None
# overall остаётся валидным средним доступных (в [0,1]), не схлопывается к 0.
assert 0.0 <= card_missing.overall <= 1.0