diff --git a/backend/app/services/forecasting/__init__.py b/backend/app/services/forecasting/__init__.py index ee7ea6fd..667e51eb 100644 --- a/backend/app/services/forecasting/__init__.py +++ b/backend/app/services/forecasting/__init__.py @@ -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", diff --git a/backend/app/services/forecasting/product_scoring.py b/backend/app/services/forecasting/product_scoring.py new file mode 100644 index 00000000..cae12b99 --- /dev/null +++ b/backend/app/services/forecasting/product_scoring.py @@ -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" diff --git a/backend/tests/services/forecasting/test_product_scoring.py b/backend/tests/services/forecasting/test_product_scoring.py new file mode 100644 index 00000000..a57d9497 --- /dev/null +++ b/backend/tests/services/forecasting/test_product_scoring.py @@ -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