backend/app/services/forecasting/__init__.py

@@ -22,6 +22,9 @@
   • product_scoring (#985/954-B)   — §14.2 десять продуктовых скоров ∈ [0,1] (выше=
     лучше) + взвешенный overall (renorm над доступными) + §16 причина на скор;
     сводит #950…#984 + live-стек, graceful-None, ADVISORY.
+  • special_indices (#986/954-C)   — §25 шесть специальных индексов (Launch Window,
+    Product Void, Cannibalization, Competitor Strength, Artificial Demand, Cost-of-
+    Error); сборка над #980/#981/§9.1/§9.2/§7.9, per-index graceful-None, ADVISORY.
 
 Источники данных:
   • макро — таблица macro_indicator через reader site_finder/macro.py (reuse).
@@ -88,6 +91,11 @@ from app.services.forecasting.scenarios import (
     build_rate_envelopes,
     compute_scenarios,
 )
+from app.services.forecasting.special_indices import (
+    SpecialIndex,
+    SpecialIndices,
+    compute_special_indices,
+)
 from app.services.forecasting.what_to_build import (
     RankedSegment,
     WhatToBuildRanking,
@@ -107,6 +115,8 @@ __all__ = [
     "SalesSeries",
     "ScenarioForecast",
     "SegmentSpec",
+    "SpecialIndex",
+    "SpecialIndices",
     "WhatToBuildRanking",
     "assemble_coefficient",
     "best_lag",
@@ -120,6 +130,7 @@ __all__ = [
     "compute_rate_sensitivity",
     "compute_scenarios",
     "compute_score_card",
+    "compute_special_indices",
     "f_issuance",
     "f_mortgage_rate",
     "f_overdue",

backend/app/services/forecasting/special_indices.py

@@ -0,0 +1,862 @@
+"""§25 шесть «специальных индексов» Site Finder v2 — ДЕТЕРМИНИРОВАННЫЙ advisory-слой.
+
+#986 (954-C, Site Finder v2 / «GG-форсайт» ТЗ §25), EPIC 10 «специальные индексы».
+Это **СБОРОЧНЫЙ слой над уже-смерженным форсайт-стеком**: своей §9.x-математики НЕ
+пересобирает — прогоняет/читает построенные сервисы (#980 demand_supply_forecast,
+#981 what_to_build, §9.1 competitors, §9.2 market_metrics, §7.9 affordability) и
+сводит их выводы в шесть продуктовых индексов §25. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM.
+
+Шесть индексов §25 (источник — переиспользуемый сервис):
+  1. Launch Window     ← скан deficit_index #980 по горизонтам {6,12,18,24}: горизонт,
+                          где дефицит ПИКОВЫЙ = рекомендованное окно запуска. SOLID.
+  2. Product Void      ← #981 rank_segments: сегменты с deficit_index ≥ _VOID_THRESHOLD
+                          (белое пятно — недообеспеченный спрос). QUICK-WIN, SOLID.
+  3. Cannibalization   ← §9.1 competitors: доля конкурентов В ТОМ ЖЕ классе, взвешенная
+                          relevance — пересечение рек. сегмента с чужим/своим стоком.
+                          DEGRADE-to-None, если нет геометрии cad_num.
+  4. Competitor Strength ← §9.1 competitors: среднее relevance_weight топ-N. DEGRADE,
+                          если нет cad_num/геометрии.
+  5. Artificial Demand ← НОВЫЙ read objective_lots: доля проданных лотов с ипотекой
+                          (encumbrance_type/bank_name) — спрос, «подпёртый» кредитом
+                          (преим. субсидированным). QUICK-WIN, SOLID; DEGRADE если
+                          нет проданных лотов (НЕ фабрикуем).
+  6. Cost-of-Error     ← §9.2 market_metrics (overstock) × средний чек (§7.9 цена ₽/м²
+                          × эталонная площадь): магнитуда убытка от неверного продукта.
+
+ADVISORY (КРИТИЧНО — честность): каждый индекс наследует advisory-статус источника
+(#980/§9.x не провалидированы до бэктеста #951; §7.9 — деградированный прокси). Поэтому
+`advisory` ВСЕГДА True и confidence жёстко ≤ _CONFIDENCE_CAP ('medium'). Цифры — для
+explainability/прототипа, НЕ основание для инвест-решения.
+
+Graceful-degrade (дух demand_supply_forecast / market_metrics): КАЖДЫЙ индекс считается
+в собственном try/except — сбой/тонкие данные одного НЕ валит карточку: индекс отдаёт
+value=None, method='unavailable', confidence='low', а остальные пять считаются. Карточка
+`SpecialIndices` возвращается ВСЕГДА (никогда не crash, никогда 0-как-заглушка).
+Детерминированно (стабильные источники + чистые helper'ы).
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import 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.sales_series import SegmentSpec
+from app.services.forecasting.what_to_build import rank_segments
+from app.services.site_finder.competitors import get_competitors
+from app.services.site_finder.market_metrics import compute_market_metrics
+
+logger = logging.getLogger(__name__)
+
+Confidence = Literal["high", "medium", "low"]
+
+# ── Названия (ключи) шести индексов §25 — стабильный публичный контракт ─────────
+KEY_LAUNCH_WINDOW = "launch_window"
+KEY_PRODUCT_VOID = "product_void"
+KEY_CANNIBALIZATION = "cannibalization"
+KEY_COMPETITOR_STRENGTH = "competitor_strength"
+KEY_ARTIFICIAL_DEMAND = "artificial_demand"
+KEY_COST_OF_ERROR = "cost_of_error"
+
+# Порядок шести индексов в карточке (детерминированный обход оркестратора).
+_INDEX_KEYS: tuple[str, ...] = (
+    KEY_LAUNCH_WINDOW,
+    KEY_PRODUCT_VOID,
+    KEY_CANNIBALIZATION,
+    KEY_COMPETITOR_STRENGTH,
+    KEY_ARTIFICIAL_DEMAND,
+    KEY_COST_OF_ERROR,
+)
+
+# ── Named-константы (вся «магия» вынесена; tunable) ───────────────────────────
+
+# Дефолтные горизонты скана Launch Window (мес) — полгода…2 года (зеркалит
+# _DEFAULT_HORIZONS #980). Окно запуска = горизонт пикового дефицита из этого набора.
+_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)
+
+# Порог deficit_index, при котором сегмент считается «белым пятном» (Product Void):
+# спрос ощутимо обгоняет предложение. 0.25 на лог-шкале #980 [−1,+1] ≈ умеренно-
+# сильный дефицит (не любой плюс). Tunable — поднять = строже определение пустоты.
+_VOID_THRESHOLD: float = 0.25
+
+# Горизонт (мес), на котором #981 ранжирует сетку для Product Void. 12 мес = типовой
+# средне-срочный продуктовый горизонт (год до выхода в продажу). Tunable.
+_VOID_HORIZON_MONTHS: int = 12
+
+# Сколько топ-конкурентов усредняем для Competitor Strength (среднее relevance_weight
+# топ-N). 5 — устойчивая оценка «силы окружения» без хвоста нерелевантных. Tunable.
+_COMPETITOR_TOP_N: int = 5
+
+# Радиус (км) для §9.1 competitors при Cannibalization/Competitor Strength. 1.0 км =
+# дефолт CompetitorsRequest (пешая доступность — реальная зона пересечения спроса).
+_COMPETITOR_RADIUS_KM: float = 1.0
+
+# Горизонт (мес) для horizon-aware relevance конкурентов (stage_at_horizon в §9.1):
+# конкурент, распроданный к нашему запуску, менее релевантен. 12 мес — дефолт.
+_COMPETITOR_HORIZON_MONTHS: int = 12
+
+# Saturation для нормализации пикового дефицита Launch Window в [0,1]: при каком
+# deficit_index сила окна достигает 1.0. 0.5 = половина шкалы #980 уже «сильное окно».
+# Линейная отсечка max(0, min(1, peak / saturation)). Tunable.
+_LAUNCH_PEAK_SATURATION: float = 0.5
+
+# Эталонная площадь (м²) для среднего чека Cost-of-Error. 50 м² зеркалит §7.9
+# _REF_AREA_M2 (медианная 1-2-к масс-рынка ЕКБ) — чек сопоставим между сегментами
+# при фиксированной площади (плавает только цена ₽/м²). Tunable.
+_COST_TICKET_AREA_M2: float = 50.0
+
+# Нормировка среднего чека (₽) в [0,1] для Cost-of-Error: чек, при котором ценовой
+# множитель достигает 1.0. 15 млн ₽ ≈ дорогой чек масс-рынка ЕКБ (50 м² × 300 тыс/м²).
+# Tunable — выше = «дорогим» считается больший чек.
+_COST_TICKET_NORM_RUB: float = 15_000_000.0
+
+# Жёсткий потолок итогового confidence: весь слой advisory (источники не валидированы
+# до бэктеста #951; §7.9 — прокси). Зеркалит _CONFIDENCE_CAP #980.
+_CONFIDENCE_CAP: Confidence = "medium"
+
+# Ранг уверенности (выше = лучше). Зеркало demand_supply_forecast/what_to_build.
+_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}
+_RANK_TO_CONFIDENCE: dict[int, Confidence] = {0: "low", 1: "medium", 2: "high"}
+
+# Метка метода у индекса, который не удалось посчитать (сбой/тонкие данные).
+_METHOD_UNAVAILABLE: str = "unavailable"
+
+# Весь слой советующий — не основание для инвест-решения.
+_ADVISORY: bool = True
+
+
+@dataclass(frozen=True)
+class SpecialIndex:
+    """Один из шести §25 индексов — нормализованное значение + объяснение. ADVISORY.
+
+    `value` ∈ [0,1] когда задан (выше = сильнее сигнал индекса), либо None при
+    тонких данных/сбое (НИКОГДА 0-как-заглушка). `label` — человекочитаемый дескриптор
+    (напр. рекомендованный горизонт Launch Window «18 мес» или RU-метка силы). `method`
+    описывает источник/способ ('unavailable' если не посчитан). `detail` — опорные
+    числа (explainability). `advisory` ВСЕГДА True.
+    """
+
+    key: str
+    value: float | None
+    label: str | None
+    confidence: Confidence  # ≤ _CONFIDENCE_CAP (advisory-cap)
+    detail: dict[str, Any]
+    method: str  # источник/способ; '_METHOD_UNAVAILABLE' если не посчитан
+    advisory: bool  # ВСЕГДА True
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "key": self.key,
+            "value": _round_or_none(self.value, 3),
+            "label": self.label,
+            "confidence": self.confidence,
+            "detail": dict(self.detail),
+            "method": self.method,
+            "advisory": self.advisory,
+        }
+
+
+@dataclass(frozen=True)
+class SpecialIndices:
+    """§25 карточка шести специальных индексов для сегмента/района/участка. ADVISORY.
+
+    `indices` содержит ВСЕ шесть ключей (_INDEX_KEYS) — даже недоступные (value=None,
+    method='unavailable'): контракт стабилен, карточка возвращается ВСЕГДА (graceful).
+    `confidence` = MIN по доступным индексам, ≤ _CONFIDENCE_CAP. `advisory` ВСЕГДА True —
+    каждый индекс наследует advisory-статус источника; карточка СОВЕТУЮЩАЯ, не основание
+    для инвест-решения.
+    """
+
+    segment: dict[str, str | None]
+    district: str | None
+    indices: dict[str, SpecialIndex]  # все 6 ключей _INDEX_KEYS
+    advisory: bool  # ВСЕГДА True
+    confidence: Confidence  # MIN по доступным, ≤ _CONFIDENCE_CAP
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "segment": dict(self.segment),
+            "district": self.district,
+            "indices": {k: v.as_dict() for k, v in self.indices.items()},
+            "advisory": self.advisory,
+            "confidence": self.confidence,
+        }
+
+
+def _round_or_none(value: float | None, digits: int) -> float | None:
+    return round(value, digits) if value is not None else None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure-математика — без БД, полностью юнит-тестируемо.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _clamp01(value: float) -> float:
+    """Зажать в [0,1]. PURE."""
+    return max(0.0, min(1.0, value))
+
+
+def _cap_confidence(confidence: Confidence, *, cap: Confidence = _CONFIDENCE_CAP) -> Confidence:
+    """Ограничить confidence сверху (advisory-слой не надёжнее cap). PURE.
+
+    Ранг-минимум(confidence, cap): high+cap-medium → medium; low → low. Зеркалит #980.
+    """
+    capped_rank = min(_CONFIDENCE_RANK[confidence], _CONFIDENCE_RANK[cap])
+    return _RANK_TO_CONFIDENCE[capped_rank]
+
+
+def _min_confidence(values: Sequence[Confidence | None]) -> Confidence:
+    """Итоговая уверенность = MIN компонентных (худшая тянет вниз). PURE.
+
+    None в списке игнорируем (компонент без сигнала). Пустой/весь-None → 'low'.
+    Зеркалит demand_supply_forecast._min_confidence.
+    """
+    ranks = [_CONFIDENCE_RANK[v] for v in values if v is not None]
+    if not ranks:
+        return "low"
+    return _RANK_TO_CONFIDENCE[min(ranks)]
+
+
+def _pick_launch_window(
+    deficit_by_horizon: dict[int, float | None],
+    *,
+    saturation: float = _LAUNCH_PEAK_SATURATION,
+) -> tuple[int | None, float | None]:
+    """Выбрать горизонт пикового дефицита и нормализованную силу окна. PURE.
+
+    Из {горизонт: deficit_index} берём горизонт с МАКСИМАЛЬНЫМ deficit_index (None-
+    горизонты игнорируются — там предложение неизмеримо). Сила окна = clamp01(peak /
+    saturation): положительный пик нормализуется в [0,1], НЕположительный пик (баланс/
+    затоварка на всех горизонтах) → 0.0 (окна «строить» нет, но горизонт-аргмакс
+    возвращаем для explainability). Все deficit None → (None, None) — деградация.
+
+    Tie-break при равных deficit_index: МЕНЬШИЙ горизонт (раньше выходить выгоднее).
+    Детерминированно.
+
+    Args:
+        deficit_by_horizon: deficit_index #980 по горизонтам (мес).
+        saturation: deficit_index, при котором сила окна = 1.0.
+
+    Returns:
+        (best_horizon | None, window_strength ∈ [0,1] | None).
+    """
+    scored = [(h, d) for h, d in deficit_by_horizon.items() if d is not None]
+    if not scored:
+        return (None, None)
+    # max по deficit, tie-break — меньший горизонт (negate deficit для max через min-ключ).
+    best_horizon, best_deficit = min(scored, key=lambda hd: (-hd[1], hd[0]))
+    if saturation <= 0:
+        strength = 1.0 if best_deficit > 0 else 0.0
+    else:
+        strength = _clamp01(best_deficit / saturation)
+    return (best_horizon, strength)
+
+
+def _void_index(deficits: Sequence[float | None], *, threshold: float = _VOID_THRESHOLD) -> float:
+    """Доля сегментов сетки с deficit_index ≥ threshold (Product Void). PURE.
+
+    «Белое пятно» = сегмент, где спрос ощутимо обгоняет предложение (deficit_index ≥
+    threshold). Индекс = доля таких сегментов среди ВСЕХ ОЦЕНЁННЫХ (deficit_index не
+    None — тонкие сегменты в знаменатель не входят, они уже отброшены #981). Пустой
+    вход / все None → 0.0 (нет измеримой пустоты, НЕ None: 0 — валидный «пустоты нет»).
+    ∈ [0,1]. PURE.
+
+    Args:
+        deficits: deficit_index ранжированных сегментов (#981 уже отбросил None-ячейки).
+        threshold: порог «белого пятна».
+
+    Returns:
+        Доля сегментов-пустот ∈ [0,1].
+    """
+    measured = [d for d in deficits if d is not None]
+    if not measured:
+        return 0.0
+    n_void = sum(1 for d in measured if d >= threshold)
+    return n_void / len(measured)
+
+
+def _count_void(deficits: Sequence[float | None], *, threshold: float = _VOID_THRESHOLD) -> int:
+    """Число сегментов-пустот (deficit_index ≥ threshold). PURE (для detail)."""
+    return sum(1 for d in deficits if d is not None and d >= threshold)
+
+
+def _competitor_strength(
+    relevance_weights: Sequence[float | None], *, top_n: int = _COMPETITOR_TOP_N
+) -> float | None:
+    """Среднее relevance_weight топ-N конкурентов (Competitor Strength). PURE.
+
+    Сортируем relevance_weight по убыванию, берём топ-N, усредняем. relevance_weight
+    уже ∈ [0,1] (детерминированный §9.1), поэтому и среднее ∈ [0,1]. None-веса
+    игнорируем. Нет конкурентов / все None → None (неизмеримо, НЕ 0 — отсутствие
+    конкурентов ≠ «нулевая сила», это деградация по данным). PURE.
+
+    Args:
+        relevance_weights: relevance_weight конкурентов (§9.1).
+        top_n: сколько верхних усреднять.
+
+    Returns:
+        Среднее топ-N ∈ [0,1] или None.
+    """
+    weights = sorted((w for w in relevance_weights if w is not None), reverse=True)
+    if not weights:
+        return None
+    top = weights[: max(1, top_n)]
+    return sum(top) / len(top)
+
+
+def _cannibalization_index(
+    same_class_relevance: Sequence[float | None],
+    all_relevance: Sequence[float | None],
+) -> float | None:
+    """Каннибализация = доля «силы» конкурентов В ТОМ ЖЕ классе. PURE.
+
+    Пересечение рекомендованного сегмента с чужим/своим стоком: чем больше суммарной
+    relevance-«силы» приходится на конкурентов В ТОМ ЖЕ классе (sum same-class /
+    sum all), тем выше риск каннибализации. relevance-веса ∈ [0,1] → доля ∈ [0,1].
+    Нет конкурентов вообще (пустой all) → None (неизмеримо — нет окружения для
+    пересечения). Конкуренты есть, но ни одного в нашем классе → 0.0 (валидно: нет
+    пересечения = нет каннибализации). PURE.
+
+    Args:
+        same_class_relevance: relevance_weight конкурентов В классе рек. сегмента.
+        all_relevance: relevance_weight ВСЕХ конкурентов в радиусе.
+
+    Returns:
+        Доля same-class-силы ∈ [0,1] или None.
+    """
+    total = sum(w for w in all_relevance if w is not None)
+    if total <= 0:
+        return None
+    same = sum(w for w in same_class_relevance if w is not None)
+    return _clamp01(same / total)
+
+
+def _artificial_demand_share(n_mortgage: int | None, n_sold: int | None) -> float | None:
+    """Доля проданных лотов, профинансированных ипотекой (Artificial Demand). PURE.
+
+    Спрос, «подпёртый» кредитом (преим. субсидированной ипотекой/спекуляцией) =
+    доля сделок с зафиксированным обременением/банком среди ВСЕХ проданных лотов.
+    Нет проданных лотов → None (неизмеримо — НЕ фабрикуем 0: отсутствие данных ≠
+    «нулевая искусственность»). ∈ [0,1] (clamp на случай грязных данных). PURE.
+
+    Args:
+        n_mortgage: проданных лотов с ипотечным признаком (encumbrance/bank).
+        n_sold: всего проданных лотов.
+
+    Returns:
+        Доля ипотечных сделок ∈ [0,1] или None.
+    """
+    if n_mortgage is None or not n_sold or n_sold <= 0:
+        return None
+    return _clamp01(float(n_mortgage) / float(n_sold))
+
+
+def _avg_ticket_rub(
+    price_per_m2: float | None, *, area_m2: float = _COST_TICKET_AREA_M2
+) -> float | None:
+    """Средний чек (₽) эталонной квартиры = цена ₽/м² × площадь. PURE.
+
+    None/непозитивная цена → None (нет чека — нечего считать). PURE.
+    """
+    if price_per_m2 is None or price_per_m2 <= 0:
+        return None
+    return price_per_m2 * area_m2
+
+
+def _cost_of_error_index(
+    oversupply_risk: float | None,
+    avg_ticket_rub: float | None,
+    *,
+    ticket_norm_rub: float = _COST_TICKET_NORM_RUB,
+) -> float | None:
+    """Cost-of-Error = риск затоварки × нормированный чек. PURE, монотонно.
+
+    Магнитуда убытка от НЕВЕРНОГО продукта: высокий риск затоварки (oversupply_risk
+    ∈ [0,1] — overstock-доля #9.2 ИЛИ магнитуда отрицательного дефицита) × дорогой
+    средний чек (нормированный в [0,1]) = дорого ошибиться. Произведение двух [0,1] →
+    [0,1]; монотонно неубывающее по каждому входу. Любой вход None → None (неизмеримо).
+    PURE.
+
+    Args:
+        oversupply_risk: риск затоварки ∈ [0,1] (overstock-доля / |negative deficit|).
+        avg_ticket_rub: средний чек сегмента (₽).
+        ticket_norm_rub: чек, при котором ценовой множитель = 1.0.
+
+    Returns:
+        Индекс стоимости ошибки ∈ [0,1] или None.
+    """
+    if oversupply_risk is None or avg_ticket_rub is None:
+        return None
+    if ticket_norm_rub <= 0:
+        ticket_factor = 1.0
+    else:
+        ticket_factor = _clamp01(avg_ticket_rub / ticket_norm_rub)
+    return _clamp01(oversupply_risk) * ticket_factor
+
+
+def _oversupply_risk_from_deficit(deficit_index: float | None) -> float | None:
+    """Риск затоварки из deficit_index #980: магнитуда ОТРИЦАТЕЛЬНОГО дефицита. PURE.
+
+    deficit_index < 0 = затоварка → риск = |deficit_index| ∈ (0,1]. deficit_index ≥ 0
+    (дефицит/баланс) → 0.0 (нет риска затоварки). None → None. Fallback-канал для
+    Cost-of-Error, когда overstock_index §9.2 недоступен. PURE.
+    """
+    if deficit_index is None:
+        return None
+    return -deficit_index if deficit_index < 0 else 0.0
+
+
+def _unavailable(key: str, *, reason: str) -> SpecialIndex:
+    """Сконструировать «недоступный» индекс (value=None, method=unavailable). PURE.
+
+    Единая форма деградации одного индекса: карточка остаётся полной (6 ключей),
+    confidence='low', detail несёт причину. advisory ВСЕГДА True.
+    """
+    return SpecialIndex(
+        key=key,
+        value=None,
+        label=None,
+        confidence="low",
+        detail={"reason": reason},
+        method=_METHOD_UNAVAILABLE,
+        advisory=_ADVISORY,
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Artificial-Demand SQL — ЕДИНСТВЕННЫЙ новый read (objective_lots). psycopg v3.
+# ──────────────────────────────────────────────────────────────────────────────
+
+# Доля ипотечных сделок среди проданных лотов сегмента. Ипотечный признак =
+# encumbrance_type ЛИБО bank_name заполнены (Объектив пишет банк/тип обременения
+# для кредитных сделок). Фильтры по district/class/premise_kind — все опциональны
+# (CAST(:x AS text) IS NULL OR ...). «Продан» = is_sold IS TRUE ИЛИ contract_date
+# заполнена (две независимые отметки сделки в objective_lots). psycopg v3: CAST,
+# НИКОГДА :x::type.
+_ARTIFICIAL_DEMAND_SQL = text(
+    """
+    WITH sold AS (
+        SELECT
+            ol.objective_lot_id,
+            (ol.encumbrance_type IS NOT NULL OR ol.bank_name IS NOT NULL) AS has_mortgage
+          FROM objective_lots ol
+         WHERE ol.premise_kind = CAST(:premise_kind AS text)
+           AND (ol.is_sold IS TRUE OR ol.contract_date IS NOT NULL)
+           AND (
+                CAST(:district AS text) IS NULL
+                OR ol.district = CAST(:district AS text)
+               )
+           AND (
+                CAST(:obj_class AS text) IS NULL
+                OR lower(ol.class) = lower(CAST(:obj_class AS text))
+               )
+    )
+    SELECT
+        COUNT(*)                                  AS n_sold,
+        COUNT(*) FILTER (WHERE has_mortgage)      AS n_mortgage
+      FROM sold
+    """
+)
+
+
+def _query_artificial_demand(
+    db: Session,
+    *,
+    district: str | None,
+    obj_class: str | None,
+    premise_kind: str,
+) -> dict[str, int]:
+    """Прочитать (n_sold, n_mortgage) сегмента из objective_lots. Graceful.
+
+    Параметризовано (psycopg v3 CAST). Сбой/пустой результат → {n_sold:0,n_mortgage:0}
+    (НЕ crash — выше преобразуется в деградацию-None). Детерминированно.
+    """
+    row = (
+        db.execute(
+            _ARTIFICIAL_DEMAND_SQL,
+            {
+                "premise_kind": premise_kind,
+                "district": district,
+                "obj_class": obj_class,
+            },
+        )
+        .mappings()
+        .first()
+    )
+    if not row:
+        return {"n_sold": 0, "n_mortgage": 0}
+    return {
+        "n_sold": int(row["n_sold"] or 0),
+        "n_mortgage": int(row["n_mortgage"] or 0),
+    }
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Per-index builders — каждый тонкий, graceful; pure-математика выше тестируется без БД.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _build_launch_window(
+    db: Session,
+    *,
+    spec: SegmentSpec,
+    district: str | None,
+    cad_num: str | None,
+    horizons: Sequence[int],
+) -> SpecialIndex:
+    """Index 1 — Launch Window: горизонт пикового deficit_index #980. SOLID/degrade."""
+    forecasts = compute_demand_supply_forecast(
+        db,
+        spec=spec,
+        district=district,
+        cad_num=cad_num or "",
+        horizons=list(horizons),
+    )
+    deficit_by_horizon = {f.horizon_months: f.deficit_index for f in forecasts}
+    confidences = [f.confidence for f in forecasts]
+    best_horizon, strength = _pick_launch_window(deficit_by_horizon)
+    if best_horizon is None:
+        return _unavailable(
+            KEY_LAUNCH_WINDOW,
+            reason="deficit_index None на всех горизонтах (тонкое предложение)",
+        )
+    confidence = _cap_confidence(_min_confidence(confidences))
+    return SpecialIndex(
+        key=KEY_LAUNCH_WINDOW,
+        value=strength,
+        label=f"{best_horizon} мес",
+        confidence=confidence,
+        detail={
+            "best_horizon_months": best_horizon,
+            "deficit_by_horizon": {h: _round_or_none(d, 3) for h, d in deficit_by_horizon.items()},
+        },
+        method="deficit_peak_scan",
+        advisory=_ADVISORY,
+    )
+
+
+def _build_product_void(
+    db: Session,
+    *,
+    district: str | None,
+    cad_num: str | None,
+    horizon_months: int,
+) -> SpecialIndex:
+    """Index 2 — Product Void: доля сегментов сетки с deficit_index ≥ порога. SOLID."""
+    ranking = rank_segments(
+        db,
+        district=district,
+        cad_num=cad_num or "",
+        horizon_months=horizon_months,
+    )
+    deficits = [seg.deficit_index for seg in ranking.ranked]
+    if not deficits:
+        return _unavailable(
+            KEY_PRODUCT_VOID,
+            reason="ранкинг #981 пуст (вся сетка тонкая)",
+        )
+    value = _void_index(deficits)
+    n_void = _count_void(deficits)
+    top_void = [seg.as_dict() for seg in ranking.ranked if seg.deficit_index >= _VOID_THRESHOLD][
+        :_COMPETITOR_TOP_N
+    ]
+    confidence = _cap_confidence(_min_confidence([seg.confidence for seg in ranking.ranked]))
+    return SpecialIndex(
+        key=KEY_PRODUCT_VOID,
+        value=value,
+        label=f"{n_void} белых пятен" if n_void else "нет белых пятен",
+        confidence=confidence,
+        detail={
+            "n_void": n_void,
+            "n_ranked": ranking.n_cells_ranked,
+            "threshold": _VOID_THRESHOLD,
+            "top_void_segments": top_void,
+        },
+        method="deficit_threshold_share",
+        advisory=_ADVISORY,
+    )
+
+
+def _fetch_competitors(db: Session, *, cad_num: str) -> list[Any]:
+    """Достать конкурентов §9.1 для cad_num (horizon-aware relevance). Может бросить.
+
+    ValueError при отсутствии геометрии участка — пробрасывается в builder, где станет
+    деградацией-None (Cannibalization/Competitor Strength требуют геометрии).
+    """
+    response = get_competitors(
+        db,
+        cad_num,
+        CompetitorsRequest(
+            radius_km=_COMPETITOR_RADIUS_KM,
+            horizon_months=_COMPETITOR_HORIZON_MONTHS,
+        ),
+    )
+    return list(response.competitors)
+
+
+def _build_competitor_strength(db: Session, *, cad_num: str | None) -> SpecialIndex:
+    """Index 4 — Competitor Strength: среднее relevance_weight топ-N. Degrade без cad_num."""
+    if not cad_num:
+        return _unavailable(
+            KEY_COMPETITOR_STRENGTH,
+            reason="нет cad_num — конкурентов §9.1 не определить",
+        )
+    competitors = _fetch_competitors(db, cad_num=cad_num)
+    weights = [c.relevance_weight for c in competitors]
+    value = _competitor_strength(weights)
+    if value is None:
+        return _unavailable(
+            KEY_COMPETITOR_STRENGTH,
+            reason="нет конкурентов с relevance_weight в радиусе",
+        )
+    return SpecialIndex(
+        key=KEY_COMPETITOR_STRENGTH,
+        value=value,
+        label=f"топ-{min(_COMPETITOR_TOP_N, len(weights))} конкурентов",
+        confidence=_cap_confidence("medium"),
+        detail={
+            "n_competitors": len(competitors),
+            "top_n": _COMPETITOR_TOP_N,
+            "radius_km": _COMPETITOR_RADIUS_KM,
+        },
+        method="relevance_weight_top_n_mean",
+        advisory=_ADVISORY,
+    )
+
+
+def _build_cannibalization(db: Session, *, spec: SegmentSpec, cad_num: str | None) -> SpecialIndex:
+    """Index 3 — Cannibalization: доля relevance-силы конкурентов В классе. Degrade."""
+    if not cad_num:
+        return _unavailable(
+            KEY_CANNIBALIZATION,
+            reason="нет cad_num/геометрии — пересечение стока не определить",
+        )
+    competitors = _fetch_competitors(db, cad_num=cad_num)
+    all_relevance = [c.relevance_weight for c in competitors]
+    target_class = (spec.obj_class or "").strip().lower()
+    same_class_relevance = [
+        c.relevance_weight
+        for c in competitors
+        if target_class and (c.obj_class or "").strip().lower() == target_class
+    ]
+    value = _cannibalization_index(same_class_relevance, all_relevance)
+    if value is None:
+        return _unavailable(
+            KEY_CANNIBALIZATION,
+            reason="нет конкурентов в радиусе — пересечения нет",
+        )
+    return SpecialIndex(
+        key=KEY_CANNIBALIZATION,
+        value=value,
+        label=spec.obj_class,
+        confidence=_cap_confidence("medium"),
+        detail={
+            "n_competitors": len(competitors),
+            "n_same_class": len(same_class_relevance),
+            "target_class": spec.obj_class,
+            "radius_km": _COMPETITOR_RADIUS_KM,
+        },
+        method="same_class_relevance_share",
+        advisory=_ADVISORY,
+    )
+
+
+def _build_artificial_demand(
+    db: Session, *, spec: SegmentSpec, district: str | None, premise_kind: str
+) -> SpecialIndex:
+    """Index 5 — Artificial Demand: доля ипотечных сделок objective_lots. SOLID/degrade."""
+    counts = _query_artificial_demand(
+        db,
+        district=district,
+        obj_class=spec.obj_class,
+        premise_kind=premise_kind,
+    )
+    n_sold = counts["n_sold"]
+    n_mortgage = counts["n_mortgage"]
+    value = _artificial_demand_share(n_mortgage, n_sold)
+    if value is None:
+        return _unavailable(
+            KEY_ARTIFICIAL_DEMAND,
+            reason="нет проданных лотов сегмента — доля ипотеки неизмерима (НЕ фабрикуем)",
+        )
+    # Уверенность по размеру выборки проданных лотов (малая выборка → low).
+    confidence: Confidence = "medium" if n_sold >= 30 else "low"
+    return SpecialIndex(
+        key=KEY_ARTIFICIAL_DEMAND,
+        value=value,
+        label=f"{round(value * 100)}% сделок с ипотекой",
+        confidence=_cap_confidence(confidence),
+        detail={
+            "n_sold": n_sold,
+            "n_mortgage": n_mortgage,
+        },
+        method="mortgage_share_objective_lots",
+        advisory=_ADVISORY,
+    )
+
+
+def _build_cost_of_error(
+    db: Session,
+    *,
+    spec: SegmentSpec,
+    district: str | None,
+    cad_num: str | None,
+    premise_kind: str,
+) -> SpecialIndex:
+    """Index 6 — Cost-of-Error: риск затоварки × средний чек. SOLID/degrade.
+
+    Риск затоварки — overstock_index §9.2 (предпочтительно); если он None — fallback
+    на магнитуду отрицательного deficit_index #980 (затоварка). Чек — сегментная цена
+    ₽/м² §7.9 (compute_affordability, reuse build_sales_series) × эталонная площадь.
+    Любой канал тонкий → None.
+    """
+    metrics = compute_market_metrics(db, district=district, premise_kind=premise_kind)
+    oversupply_risk = metrics.overstock_index
+    risk_source = "overstock_index"
+
+    if oversupply_risk is None and cad_num:
+        # Fallback-канал: магнитуда отрицательного дефицита #980 на дефолтном горизонте.
+        forecasts = compute_demand_supply_forecast(
+            db,
+            spec=spec,
+            district=district,
+            cad_num=cad_num,
+            horizons=[_VOID_HORIZON_MONTHS],
+        )
+        if forecasts:
+            oversupply_risk = _oversupply_risk_from_deficit(forecasts[0].deficit_index)
+            risk_source = "negative_deficit"
+
+    # Сегментная цена ₽/м² — из §7.9 (reuse build_sales_series внутри); ref-площадь общая.
+    price_per_m2 = compute_affordability(
+        db, spec=spec, ref_area_m2=_COST_TICKET_AREA_M2
+    ).price_per_m2
+    avg_ticket = _avg_ticket_rub(price_per_m2)
+    value = _cost_of_error_index(oversupply_risk, avg_ticket)
+    if value is None:
+        return _unavailable(
+            KEY_COST_OF_ERROR,
+            reason="нет риска затоварки или среднего чека (тонкие §9.2/§7.9)",
+        )
+    return SpecialIndex(
+        key=KEY_COST_OF_ERROR,
+        value=value,
+        label="риск дорогой ошибки",
+        confidence=_cap_confidence(metrics.confidence),
+        detail={
+            "oversupply_risk": _round_or_none(oversupply_risk, 3),
+            "risk_source": risk_source,
+            "avg_ticket_rub": _round_or_none(avg_ticket, 0),
+            "price_per_m2": _round_or_none(price_per_m2, 0),
+            "ref_area_m2": _COST_TICKET_AREA_M2,
+        },
+        method="oversupply_x_ticket",
+        advisory=_ADVISORY,
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# DB-оркестратор — каждый индекс в своём try/except; карточка возвращается ВСЕГДА.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def compute_special_indices(
+    db: Session,
+    *,
+    spec: SegmentSpec,
+    district: str | None = None,
+    cad_num: str | None = None,
+    horizons: Sequence[int] = _DEFAULT_HORIZONS,
+    premise_kind: str = "квартира",
+) -> SpecialIndices:
+    """§25 собрать карточку шести специальных индексов для сегмента/района/участка.
+
+    ADVISORY (каждый индекс наследует advisory-статус источника, см. module docstring) —
+    НЕ подключать в production-эндпоинт как основание для решения. СБОРКА над
+    форсайт-стеком, НЕ пересчёт §9.x.
+
+    Считает шесть индексов §25, КАЖДЫЙ в собственном try/except: сбой/тонкие данные
+    одного индекса → этот индекс отдаёт value=None + method='unavailable' (деградация),
+    а остальные пять считаются. Карточка возвращается ВСЕГДА (graceful, никогда crash).
+    `advisory` ВСЕГДА True; `confidence` = MIN по доступным индексам, ≤ _CONFIDENCE_CAP.
+
+    Quick-win SOLID (данные есть): Product Void (#981), Artificial Demand (ипотечный
+    признак objective_lots). Остальные деградируют изящно при тонких данных: Launch
+    Window — если deficit None на всех горизонтах; Cannibalization / Competitor
+    Strength — если нет cad_num/геометрии; Cost-of-Error — если тонкие §9.2/#980.
+
+    Args:
+        db: SQLAlchemy sync Session.
+        spec: целевой (рекомендованный) сегмент рынка.
+        district: район участка (None → ЕКБ-wide).
+        cad_num: кадастровый номер участка (None → Cannibalization/Competitor Strength
+            деградируют; Launch Window/Cost-of-Error работают без конкурентов).
+        horizons: горизонты скана Launch Window (по умолчанию _DEFAULT_HORIZONS).
+        premise_kind: тип помещения (по умолчанию 'квартира').
+
+    Returns:
+        SpecialIndices — карточка с ВСЕМИ шестью ключами (всегда; тонкие → None+unavail).
+    """
+    segment = spec.as_dict()
+
+    # (ключ, нулевой-аргумент builder) — каждый вызывается в общем try/except ниже.
+    builders: dict[str, Any] = {
+        KEY_LAUNCH_WINDOW: lambda: _build_launch_window(
+            db, spec=spec, district=district, cad_num=cad_num, horizons=horizons
+        ),
+        KEY_PRODUCT_VOID: lambda: _build_product_void(
+            db, district=district, cad_num=cad_num, horizon_months=_VOID_HORIZON_MONTHS
+        ),
+        KEY_CANNIBALIZATION: lambda: _build_cannibalization(db, spec=spec, cad_num=cad_num),
+        KEY_COMPETITOR_STRENGTH: lambda: _build_competitor_strength(db, cad_num=cad_num),
+        KEY_ARTIFICIAL_DEMAND: lambda: _build_artificial_demand(
+            db, spec=spec, district=district, premise_kind=premise_kind
+        ),
+        KEY_COST_OF_ERROR: lambda: _build_cost_of_error(
+            db, spec=spec, district=district, cad_num=cad_num, premise_kind=premise_kind
+        ),
+    }
+
+    indices: dict[str, SpecialIndex] = {}
+    for key in _INDEX_KEYS:
+        try:
+            indices[key] = builders[key]()
+        except Exception:
+            # Сбой одного индекса НЕ валит карточку: деградация-None, остальные считаются.
+            logger.exception(
+                "special_indices: index=%s failed (segment=%s district=%s cad_num=%s)",
+                key,
+                segment,
+                district,
+                cad_num,
+            )
+            indices[key] = _unavailable(key, reason="ошибка расчёта (см. логи)")
+
+    confidence = _min_confidence([idx.confidence for idx in indices.values()])
+
+    n_available = sum(1 for idx in indices.values() if idx.value is not None)
+    logger.info(
+        "special_indices: segment=%s district=%s cad_num=%s available=%d/6 "
+        "confidence=%s (ADVISORY)",
+        segment,
+        district,
+        cad_num,
+        n_available,
+        confidence,
+    )
+
+    return SpecialIndices(
+        segment=segment,
+        district=district,
+        indices=indices,
+        advisory=_ADVISORY,
+        confidence=confidence,
+    )

backend/tests/services/forecasting/test_special_indices.py

@@ -0,0 +1,879 @@
+"""Тесты §25 special_indices (#986/954-C) — pure-helpers + mocked-БД + оркестратор.
+
+Покрытие:
+  • Pure-математика БЕЗ БД: Product Void (порог/доля/счёт), Launch Window (пик +
+    tie-break + нормализация), Competitor Strength (топ-N среднее), Cannibalization
+    (доля same-class), Artificial Demand (доля ипотеки), Cost-of-Error (монотонность
+    + произведение), нормализации в [0,1], confidence-helpers, _avg_ticket.
+  • Artificial-Demand SQL: MagicMock-сессия — форма SQL (CAST(:x AS type), не ::),
+    параметры; сигнал есть → индекс; нет проданных → None + caveat.
+  • compute_special_indices: @patch бэкенд-сервисов — 6 индексов присутствуют,
+    advisory True, per-index graceful (сбой одного → unavailable, карточка цела).
+
+Всё детерминировано, без БД (сессия мокается). Зеркалит стиль test_what_to_build /
+test_market_metrics.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.services.forecasting.sales_series import SegmentSpec
+from app.services.forecasting.special_indices import (
+    _INDEX_KEYS,
+    _METHOD_UNAVAILABLE,
+    KEY_ARTIFICIAL_DEMAND,
+    KEY_CANNIBALIZATION,
+    KEY_COMPETITOR_STRENGTH,
+    KEY_COST_OF_ERROR,
+    KEY_LAUNCH_WINDOW,
+    KEY_PRODUCT_VOID,
+    _artificial_demand_share,
+    _avg_ticket_rub,
+    _cannibalization_index,
+    _cap_confidence,
+    _clamp01,
+    _competitor_strength,
+    _cost_of_error_index,
+    _count_void,
+    _min_confidence,
+    _oversupply_risk_from_deficit,
+    _pick_launch_window,
+    _query_artificial_demand,
+    _void_index,
+    compute_special_indices,
+)
+
+# Patch-таргеты — имена, импортированные В модуль special_indices. _DSF используется
+# точечно (re-patch поверх _full_stack_patch); остальные сервисы патчатся через
+# patch.multiple(_MOD, ...) по kwargs-имени, поэтому отдельных констант им не нужно.
+_MOD = "app.services.forecasting.special_indices"
+_DSF = f"{_MOD}.compute_demand_supply_forecast"
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: _clamp01 / confidence-helpers
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestClamp01:
+    def test_in_range_passthrough(self) -> None:
+        assert _clamp01(0.4) == 0.4
+
+    def test_above_one_clamped(self) -> None:
+        assert _clamp01(1.7) == 1.0
+
+    def test_below_zero_clamped(self) -> None:
+        assert _clamp01(-0.3) == 0.0
+
+
+class TestCapConfidence:
+    def test_high_capped_to_medium(self) -> None:
+        assert _cap_confidence("high") == "medium"
+
+    def test_medium_unchanged(self) -> None:
+        assert _cap_confidence("medium") == "medium"
+
+    def test_low_unchanged(self) -> None:
+        assert _cap_confidence("low") == "low"
+
+
+class TestMinConfidence:
+    def test_worst_wins(self) -> None:
+        assert _min_confidence(["high", "low", "medium"]) == "low"
+
+    def test_none_ignored(self) -> None:
+        assert _min_confidence(["medium", None, "high"]) == "medium"
+
+    def test_all_none_low(self) -> None:
+        assert _min_confidence([None, None]) == "low"
+
+    def test_empty_low(self) -> None:
+        assert _min_confidence([]) == "low"
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: Product Void (порог / доля / счёт)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestVoidIndex:
+    def test_share_above_threshold(self) -> None:
+        # 2 из 4 измеренных ≥ 0.25 → 0.5.
+        assert _void_index([0.3, 0.1, 0.9, -0.2]) == 0.5
+
+    def test_threshold_is_inclusive(self) -> None:
+        # ровно на пороге 0.25 → считается пустотой.
+        assert _void_index([0.25]) == 1.0
+
+    def test_below_threshold_excluded(self) -> None:
+        # 0.24 < 0.25 → не пустота.
+        assert _void_index([0.24]) == 0.0
+
+    def test_none_cells_excluded_from_denominator(self) -> None:
+        # None не входит ни в числитель, ни в знаменатель: 1 из 2 измеренных.
+        assert _void_index([0.9, None, 0.0]) == 0.5
+
+    def test_empty_yields_zero(self) -> None:
+        assert _void_index([]) == 0.0
+
+    def test_all_none_yields_zero(self) -> None:
+        assert _void_index([None, None]) == 0.0
+
+    def test_in_range(self) -> None:
+        v = _void_index([0.9, 0.8, 0.7])
+        assert 0.0 <= v <= 1.0
+        assert v == 1.0
+
+    def test_custom_threshold(self) -> None:
+        assert _void_index([0.5, 0.4], threshold=0.45) == 0.5
+
+
+class TestCountVoid:
+    def test_counts_at_or_above_threshold(self) -> None:
+        assert _count_void([0.3, 0.25, 0.1, 0.99]) == 3
+
+    def test_none_skipped(self) -> None:
+        assert _count_void([None, 0.9, None]) == 1
+
+    def test_empty_zero(self) -> None:
+        assert _count_void([]) == 0
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: Launch Window (пик + tie-break + нормализация)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestPickLaunchWindow:
+    def test_picks_peak_horizon(self) -> None:
+        h, strength = _pick_launch_window({6: 0.1, 12: 0.4, 18: 0.2, 24: -0.1})
+        assert h == 12
+        # 0.4 / 0.5 (saturation) = 0.8.
+        assert strength == pytest.approx(0.8)
+
+    def test_strength_in_range(self) -> None:
+        _, strength = _pick_launch_window({6: 0.9, 12: 0.95})
+        assert strength is not None
+        assert 0.0 <= strength <= 1.0
+        assert strength == 1.0  # 0.95/0.5 clamp → 1.0
+
+    def test_tie_break_prefers_earlier_horizon(self) -> None:
+        # равный пиковый deficit на 6 и 18 → берём меньший горизонт (раньше выгоднее).
+        h, _ = _pick_launch_window({6: 0.5, 18: 0.5})
+        assert h == 6
+
+    def test_none_horizons_ignored(self) -> None:
+        # 12 — None (тонко), пик среди измеренных = 24.
+        h, _ = _pick_launch_window({6: 0.1, 12: None, 24: 0.3})
+        assert h == 24
+
+    def test_all_none_degrades(self) -> None:
+        assert _pick_launch_window({6: None, 12: None}) == (None, None)
+
+    def test_empty_degrades(self) -> None:
+        assert _pick_launch_window({}) == (None, None)
+
+    def test_nonpositive_peak_zero_strength_but_horizon_returned(self) -> None:
+        # все горизонты ≤0 (баланс/затоварка): окна «строить» нет (strength 0),
+        # но аргмакс-горизонт возвращаем (наименее плохой) для explainability.
+        h, strength = _pick_launch_window({6: -0.3, 12: -0.1, 24: -0.5})
+        assert h == 12  # наибольший (наименее отрицательный) deficit
+        assert strength == 0.0
+
+    def test_saturation_nonpositive_degrades_to_sign(self) -> None:
+        h, strength = _pick_launch_window({6: 0.2}, saturation=0.0)
+        assert h == 6
+        assert strength == 1.0
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: Competitor Strength (топ-N среднее)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestCompetitorStrength:
+    def test_mean_of_top_n(self) -> None:
+        # топ-5 по убыванию из 6: 0.9,0.8,0.7,0.6,0.5 → mean 0.7.
+        v = _competitor_strength([0.1, 0.5, 0.9, 0.6, 0.8, 0.7])
+        assert v == pytest.approx(0.7)
+
+    def test_fewer_than_top_n(self) -> None:
+        v = _competitor_strength([0.4, 0.6])
+        assert v == pytest.approx(0.5)
+
+    def test_none_weights_skipped(self) -> None:
+        v = _competitor_strength([None, 0.8, None, 0.4])
+        assert v == pytest.approx(0.6)
+
+    def test_empty_is_none(self) -> None:
+        assert _competitor_strength([]) is None
+
+    def test_all_none_is_none(self) -> None:
+        assert _competitor_strength([None, None]) is None
+
+    def test_in_range(self) -> None:
+        v = _competitor_strength([0.9, 0.95, 1.0])
+        assert v is not None and 0.0 <= v <= 1.0
+
+    def test_custom_top_n(self) -> None:
+        v = _competitor_strength([0.9, 0.8, 0.1], top_n=2)
+        assert v == pytest.approx(0.85)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: Cannibalization (доля same-class)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestCannibalizationIndex:
+    def test_share_of_same_class(self) -> None:
+        # same = 0.6+0.4 = 1.0; all = 0.6+0.4+0.5+0.5 = 2.0 → 0.5.
+        v = _cannibalization_index([0.6, 0.4], [0.6, 0.4, 0.5, 0.5])
+        assert v == pytest.approx(0.5)
+
+    def test_no_same_class_is_zero(self) -> None:
+        # конкуренты есть, но ни одного в нашем классе → 0.0 (валидно).
+        v = _cannibalization_index([], [0.5, 0.5])
+        assert v == 0.0
+
+    def test_no_competitors_is_none(self) -> None:
+        # окружения нет вообще → None (неизмеримо).
+        assert _cannibalization_index([], []) is None
+
+    def test_all_same_class_is_one(self) -> None:
+        v = _cannibalization_index([0.5, 0.5], [0.5, 0.5])
+        assert v == 1.0
+
+    def test_none_weights_skipped(self) -> None:
+        v = _cannibalization_index([0.5], [0.5, None, 0.5])
+        assert v == pytest.approx(0.5)
+
+    def test_in_range(self) -> None:
+        v = _cannibalization_index([0.3], [0.3, 0.7])
+        assert v is not None and 0.0 <= v <= 1.0
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: Artificial Demand (доля ипотеки)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestArtificialDemandShare:
+    def test_basic_share(self) -> None:
+        assert _artificial_demand_share(30, 100) == pytest.approx(0.3)
+
+    def test_full_mortgage(self) -> None:
+        assert _artificial_demand_share(50, 50) == 1.0
+
+    def test_zero_mortgage(self) -> None:
+        assert _artificial_demand_share(0, 80) == 0.0
+
+    def test_no_sold_is_none(self) -> None:
+        # нет проданных лотов → None (НЕ фабрикуем 0).
+        assert _artificial_demand_share(0, 0) is None
+
+    def test_none_mortgage_is_none(self) -> None:
+        assert _artificial_demand_share(None, 50) is None
+
+    def test_clamped_to_one_on_dirty_data(self) -> None:
+        # грязные данные (mortgage > sold) → clamp 1.0, не >1.
+        assert _artificial_demand_share(120, 100) == 1.0
+
+    def test_in_range(self) -> None:
+        v = _artificial_demand_share(17, 23)
+        assert v is not None and 0.0 <= v <= 1.0
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure: Cost-of-Error (монотонность + произведение + чек)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class TestAvgTicket:
+    def test_price_times_area(self) -> None:
+        # 200_000 ₽/м² × 50 м² = 10 млн.
+        assert _avg_ticket_rub(200_000.0) == 10_000_000.0
+
+    def test_none_price_is_none(self) -> None:
+        assert _avg_ticket_rub(None) is None
+
+    def test_nonpositive_price_is_none(self) -> None:
+        assert _avg_ticket_rub(0.0) is None
+
+    def test_custom_area(self) -> None:
+        assert _avg_ticket_rub(100_000.0, area_m2=80.0) == 8_000_000.0
+
+
+class TestOversupplyRiskFromDeficit:
+    def test_negative_deficit_is_magnitude(self) -> None:
+        assert _oversupply_risk_from_deficit(-0.4) == pytest.approx(0.4)
+
+    def test_positive_deficit_zero_risk(self) -> None:
+        assert _oversupply_risk_from_deficit(0.6) == 0.0
+
+    def test_zero_deficit_zero_risk(self) -> None:
+        assert _oversupply_risk_from_deficit(0.0) == 0.0
+
+    def test_none_is_none(self) -> None:
+        assert _oversupply_risk_from_deficit(None) is None
+
+
+class TestCostOfErrorIndex:
+    def test_product_of_risk_and_ticket(self) -> None:
+        # risk 0.5 × (10млн/15млн clamp=0.6667) ≈ 0.3333.
+        v = _cost_of_error_index(0.5, 10_000_000.0)
+        assert v == pytest.approx(0.5 * (10_000_000.0 / 15_000_000.0))
+
+    def test_monotonic_in_risk(self) -> None:
+        low = _cost_of_error_index(0.2, 10_000_000.0)
+        high = _cost_of_error_index(0.8, 10_000_000.0)
+        assert low is not None and high is not None
+        assert high > low
+
+    def test_monotonic_in_ticket(self) -> None:
+        cheap = _cost_of_error_index(0.5, 5_000_000.0)
+        pricey = _cost_of_error_index(0.5, 12_000_000.0)
+        assert cheap is not None and pricey is not None
+        assert pricey > cheap
+
+    def test_in_range(self) -> None:
+        v = _cost_of_error_index(0.9, 30_000_000.0)  # ticket clamps to 1.0
+        assert v is not None and 0.0 <= v <= 1.0
+        assert v == pytest.approx(0.9)
+
+    def test_none_risk_is_none(self) -> None:
+        assert _cost_of_error_index(None, 10_000_000.0) is None
+
+    def test_none_ticket_is_none(self) -> None:
+        assert _cost_of_error_index(0.5, None) is None
+
+    def test_nonpositive_norm_degrades_to_risk_only(self) -> None:
+        v = _cost_of_error_index(0.4, 10_000_000.0, ticket_norm_rub=0.0)
+        assert v == pytest.approx(0.4)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Artificial-Demand SQL: MagicMock-сессия (форма SQL + параметры + сигнал/деградация)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _mock_db_one(row: dict[str, Any] | None) -> MagicMock:
+    """Сессия, чей единственный execute().mappings().first() вернёт row."""
+    db = MagicMock()
+    result = MagicMock()
+    result.mappings.return_value.first.return_value = row
+    db.execute.return_value = result
+    return db
+
+
+def _executed_sql(db: MagicMock, call_index: int = 0) -> str:
+    args, _ = db.execute.call_args_list[call_index]
+    return str(args[0])
+
+
+def _executed_params(db: MagicMock, call_index: int = 0) -> dict[str, Any]:
+    args, _ = db.execute.call_args_list[call_index]
+    return args[1]
+
+
+class TestArtificialDemandSQL:
+    def test_sql_uses_cast_not_double_colon(self) -> None:
+        db = _mock_db_one({"n_sold": 10, "n_mortgage": 4})
+        _query_artificial_demand(
+            db, district="Академический", obj_class="комфорт", premise_kind="квартира"
+        )
+        sql = _executed_sql(db)
+        assert "CAST(:premise_kind AS text)" in sql
+        assert "CAST(:district AS text)" in sql
+        assert "CAST(:obj_class AS text)" in sql
+        # psycopg v3: никогда :x::type
+        assert ":premise_kind::" not in sql
+        assert ":district::" not in sql
+        assert ":obj_class::" not in sql
+
+    def test_sql_reads_mortgage_signal_columns(self) -> None:
+        db = _mock_db_one({"n_sold": 10, "n_mortgage": 4})
+        _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
+        sql = _executed_sql(db)
+        # сигнал ипотеки = encumbrance_type / bank_name (реальные колонки objective_lots).
+        assert "encumbrance_type" in sql
+        assert "bank_name" in sql
+        assert "objective_lots" in sql
+
+    def test_params_passed(self) -> None:
+        db = _mock_db_one({"n_sold": 1, "n_mortgage": 0})
+        _query_artificial_demand(
+            db, district="Пионерский", obj_class="бизнес", premise_kind="квартира"
+        )
+        p = _executed_params(db)
+        assert p["district"] == "Пионерский"
+        assert p["obj_class"] == "бизнес"
+        assert p["premise_kind"] == "квартира"
+
+    def test_signal_present_returns_counts(self) -> None:
+        db = _mock_db_one({"n_sold": 80, "n_mortgage": 52})
+        out = _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
+        assert out == {"n_sold": 80, "n_mortgage": 52}
+
+    def test_empty_row_degrades_to_zeros(self) -> None:
+        db = _mock_db_one(None)
+        out = _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
+        assert out == {"n_sold": 0, "n_mortgage": 0}
+
+    def test_null_counts_coerced_to_zero(self) -> None:
+        db = _mock_db_one({"n_sold": None, "n_mortgage": None})
+        out = _query_artificial_demand(db, district=None, obj_class=None, premise_kind="квартира")
+        assert out == {"n_sold": 0, "n_mortgage": 0}
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Artificial-Demand builder via mocked rows: signal → index; no signal → None+caveat
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+_SPEC = SegmentSpec(obj_class="комфорт", room_bucket="2-к 45-60", district="Академический")
+
+
+def _patch_all_unavailable() -> Any:
+    """Контекст: все 5 НЕ-Artificial бэкенда брошены/пусты (изолируем Artificial)."""
+    return patch.multiple(
+        _MOD,
+        compute_demand_supply_forecast=MagicMock(return_value=[]),
+        rank_segments=MagicMock(return_value=MagicMock(ranked=[], n_cells_ranked=0)),
+        get_competitors=MagicMock(side_effect=ValueError("no geom")),
+        compute_market_metrics=MagicMock(
+            return_value=MagicMock(overstock_index=None, confidence="low")
+        ),
+        compute_affordability=MagicMock(return_value=MagicMock(price_per_m2=None)),
+    )
+
+
+class TestArtificialDemandBuilder:
+    def test_signal_present_yields_index(self) -> None:
+        db = MagicMock()
+        with (
+            _patch_all_unavailable(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 65},
+            ),
+        ):
+            card = compute_special_indices(db, spec=_SPEC, district="Академический")
+        idx = card.indices[KEY_ARTIFICIAL_DEMAND]
+        assert idx.value == pytest.approx(0.65)
+        assert idx.method == "mortgage_share_objective_lots"
+        assert idx.detail["n_sold"] == 100
+        assert idx.detail["n_mortgage"] == 65
+        assert idx.advisory is True
+
+    def test_no_signal_yields_none_with_caveat(self) -> None:
+        db = MagicMock()
+        with (
+            _patch_all_unavailable(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 0, "n_mortgage": 0},
+            ),
+        ):
+            card = compute_special_indices(db, spec=_SPEC, district="Академический")
+        idx = card.indices[KEY_ARTIFICIAL_DEMAND]
+        assert idx.value is None
+        assert idx.method == _METHOD_UNAVAILABLE
+        assert "reason" in idx.detail
+        assert "фабрикуем" in idx.detail["reason"]  # явный caveat «не фабрикуем»
+
+    def test_small_sample_low_confidence(self) -> None:
+        db = MagicMock()
+        with (
+            _patch_all_unavailable(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 5, "n_mortgage": 3},
+            ),
+        ):
+            card = compute_special_indices(db, spec=_SPEC, district="Академический")
+        idx = card.indices[KEY_ARTIFICIAL_DEMAND]
+        assert idx.value == pytest.approx(0.6)
+        assert idx.confidence == "low"  # n_sold < 30
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# compute_special_indices: @patch бэкенда — 6 индексов, advisory, graceful per-index
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _forecast_stub(deficit: float | None, *, horizon: int, confidence: str = "medium") -> MagicMock:
+    f = MagicMock()
+    f.deficit_index = deficit
+    f.horizon_months = horizon
+    f.confidence = confidence
+    return f
+
+
+def _ranked_stub(deficit: float, *, confidence: str = "medium") -> MagicMock:
+    seg = MagicMock()
+    seg.deficit_index = deficit
+    seg.confidence = confidence
+    seg.as_dict.return_value = {"deficit_index": deficit}
+    return seg
+
+
+def _competitor_stub(relevance: float | None, obj_class: str | None) -> MagicMock:
+    c = MagicMock()
+    c.relevance_weight = relevance
+    c.obj_class = obj_class
+    return c
+
+
+def _full_stack_patch() -> Any:
+    """Все 5 бэкенд-сервисов отдают здоровые данные → все 6 индексов считаются."""
+    forecasts = [_forecast_stub(0.1 * h, horizon=h) for h in (6, 12, 18, 24)]
+    ranking = MagicMock()
+    ranking.ranked = [_ranked_stub(0.5), _ranked_stub(0.3), _ranked_stub(-0.1)]
+    ranking.n_cells_ranked = 3
+    competitors_resp = MagicMock()
+    competitors_resp.competitors = [
+        _competitor_stub(0.8, "комфорт"),
+        _competitor_stub(0.6, "комфорт"),
+        _competitor_stub(0.5, "бизнес"),
+    ]
+    metrics = MagicMock()
+    metrics.overstock_index = 0.4
+    metrics.confidence = "medium"
+    afford = MagicMock()
+    afford.price_per_m2 = 200_000.0
+    return patch.multiple(
+        _MOD,
+        compute_demand_supply_forecast=MagicMock(return_value=forecasts),
+        rank_segments=MagicMock(return_value=ranking),
+        get_competitors=MagicMock(return_value=competitors_resp),
+        compute_market_metrics=MagicMock(return_value=metrics),
+        compute_affordability=MagicMock(return_value=afford),
+    )
+
+
+class TestComputeSpecialIndicesShape:
+    def test_all_six_indices_present(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        assert set(card.indices.keys()) == set(_INDEX_KEYS)
+        assert len(card.indices) == 6
+
+    def test_advisory_always_true(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        assert card.advisory is True
+        assert all(idx.advisory is True for idx in card.indices.values())
+
+    def test_all_values_in_range_when_present(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        for idx in card.indices.values():
+            if idx.value is not None:
+                assert 0.0 <= idx.value <= 1.0, idx.key
+
+    def test_confidence_capped_at_medium(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        assert card.confidence in ("low", "medium")  # никогда high
+        for idx in card.indices.values():
+            assert idx.confidence in ("low", "medium")
+
+    def test_as_dict_roundtrip(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        d = card.as_dict()
+        assert d["advisory"] is True
+        assert set(d["indices"].keys()) == set(_INDEX_KEYS)
+        assert d["district"] == "Академический"
+        # каждый индекс сериализуется с ключами контракта.
+        for sub in d["indices"].values():
+            assert set(sub.keys()) == {
+                "key",
+                "value",
+                "label",
+                "confidence",
+                "detail",
+                "method",
+                "advisory",
+            }
+
+
+class TestComputeSpecialIndicesValues:
+    def test_launch_window_picks_peak_horizon(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        lw = card.indices[KEY_LAUNCH_WINDOW]
+        # deficit = 0.1*h → пик на h=24 (0.1*24=2.4 clamps), label «24 мес».
+        assert lw.label == "24 мес"
+        assert lw.detail["best_horizon_months"] == 24
+
+    def test_product_void_counts_segments(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        pv = card.indices[KEY_PRODUCT_VOID]
+        # ranked deficits [0.5,0.3,-0.1]: 2 ≥ 0.25 → share 2/3.
+        assert pv.value == pytest.approx(2 / 3)
+        assert pv.detail["n_void"] == 2
+
+    def test_competitor_strength_mean_top_n(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        cs = card.indices[KEY_COMPETITOR_STRENGTH]
+        # relevance [0.8,0.6,0.5] → mean 0.6333.
+        assert cs.value == pytest.approx((0.8 + 0.6 + 0.5) / 3)
+
+    def test_cannibalization_same_class_share(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        can = card.indices[KEY_CANNIBALIZATION]
+        # same-class (комфорт) = 0.8+0.6=1.4; all=1.9 → 0.7368.
+        assert can.value == pytest.approx((0.8 + 0.6) / (0.8 + 0.6 + 0.5))
+
+    def test_cost_of_error_product(self) -> None:
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        coe = card.indices[KEY_COST_OF_ERROR]
+        # risk 0.4 × ticket(200k×50=10млн / 15млн = 0.6667) ≈ 0.2667.
+        ticket_factor = (200_000.0 * 50.0) / 15_000_000.0
+        assert coe.value == pytest.approx(0.4 * ticket_factor)
+        assert coe.detail["risk_source"] == "overstock_index"
+
+
+class TestComputeSpecialIndicesGraceful:
+    def test_no_cad_num_degrades_competitor_indices(self) -> None:
+        # без cad_num: Cannibalization + Competitor Strength → unavailable, остальные ок.
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(db, spec=_SPEC, district="Академический")
+        assert card.indices[KEY_CANNIBALIZATION].method == _METHOD_UNAVAILABLE
+        assert card.indices[KEY_COMPETITOR_STRENGTH].method == _METHOD_UNAVAILABLE
+        assert card.indices[KEY_CANNIBALIZATION].value is None
+        # quick-win Artificial Demand всё равно посчитан.
+        assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.4)
+
+    def test_per_index_failure_isolated(self) -> None:
+        # один сервис (rank_segments) бросает → Product Void unavailable, остальные 5 ок.
+        forecasts = [_forecast_stub(0.2, horizon=h) for h in (6, 12, 18, 24)]
+        metrics = MagicMock(overstock_index=0.3, confidence="medium")
+        afford = MagicMock(price_per_m2=180_000.0)
+        comp = MagicMock()
+        comp.competitors = [_competitor_stub(0.7, "комфорт")]
+        db = MagicMock()
+        with (
+            patch.multiple(
+                _MOD,
+                compute_demand_supply_forecast=MagicMock(return_value=forecasts),
+                rank_segments=MagicMock(side_effect=RuntimeError("boom")),
+                get_competitors=MagicMock(return_value=comp),
+                compute_market_metrics=MagicMock(return_value=metrics),
+                compute_affordability=MagicMock(return_value=afford),
+            ),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 50, "n_mortgage": 20},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        # сбойный индекс — unavailable; карточка цела (все 6 ключей).
+        assert card.indices[KEY_PRODUCT_VOID].method == _METHOD_UNAVAILABLE
+        assert card.indices[KEY_PRODUCT_VOID].value is None
+        assert len(card.indices) == 6
+        # остальные посчитаны.
+        assert card.indices[KEY_LAUNCH_WINDOW].value is not None
+        assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.4)
+        assert card.advisory is True
+
+    def test_all_backends_fail_card_still_returns(self) -> None:
+        # тотальный сбой: каждый индекс unavailable, но карточка возвращается (не crash).
+        db = MagicMock()
+        with (
+            patch.multiple(
+                _MOD,
+                compute_demand_supply_forecast=MagicMock(side_effect=RuntimeError("x")),
+                rank_segments=MagicMock(side_effect=RuntimeError("x")),
+                get_competitors=MagicMock(side_effect=RuntimeError("x")),
+                compute_market_metrics=MagicMock(side_effect=RuntimeError("x")),
+                compute_affordability=MagicMock(side_effect=RuntimeError("x")),
+            ),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                side_effect=RuntimeError("x"),
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        assert len(card.indices) == 6
+        assert all(idx.method == _METHOD_UNAVAILABLE for idx in card.indices.values())
+        assert all(idx.value is None for idx in card.indices.values())
+        assert card.advisory is True
+        assert card.confidence == "low"
+
+    def test_launch_window_all_none_degrades(self) -> None:
+        # deficit None на всех горизонтах → Launch Window unavailable.
+        forecasts = [_forecast_stub(None, horizon=h) for h in (6, 12, 18, 24)]
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(_DSF, return_value=forecasts),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        assert card.indices[KEY_LAUNCH_WINDOW].method == _METHOD_UNAVAILABLE
+        assert card.indices[KEY_LAUNCH_WINDOW].value is None
+
+    def test_cost_of_error_falls_back_to_negative_deficit(self) -> None:
+        # overstock_index None → Cost-of-Error берёт магнитуду отрицательного дефицита.
+        metrics = MagicMock(overstock_index=None, confidence="medium")
+        afford = MagicMock(price_per_m2=200_000.0)
+        # forecast на _VOID_HORIZON_MONTHS=12 (fallback) с отрицательным дефицитом.
+        forecasts = [_forecast_stub(-0.5, horizon=12)]
+        comp = MagicMock()
+        comp.competitors = [_competitor_stub(0.7, "комфорт")]
+        db = MagicMock()
+        with (
+            patch.multiple(
+                _MOD,
+                compute_demand_supply_forecast=MagicMock(return_value=forecasts),
+                rank_segments=MagicMock(return_value=MagicMock(ranked=[], n_cells_ranked=0)),
+                get_competitors=MagicMock(return_value=comp),
+                compute_market_metrics=MagicMock(return_value=metrics),
+                compute_affordability=MagicMock(return_value=afford),
+            ),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 50, "n_mortgage": 20},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        coe = card.indices[KEY_COST_OF_ERROR]
+        assert coe.detail["risk_source"] == "negative_deficit"
+        # risk |−0.5|=0.5 × ticket(10млн/15млн=0.6667) ≈ 0.3333.
+        assert coe.value == pytest.approx(0.5 * ((200_000.0 * 50.0) / 15_000_000.0))
+
+    def test_aggregate_spec_no_class_param_is_none(self) -> None:
+        # spec без obj_class → Artificial-Demand SQL получает obj_class=None (агрегат).
+        db = MagicMock()
+        captured: dict[str, Any] = {}
+
+        def _capture(_db: Any, *, district: Any, obj_class: Any, premise_kind: Any) -> dict:
+            captured["obj_class"] = obj_class
+            captured["district"] = district
+            return {"n_sold": 40, "n_mortgage": 10}
+
+        spec_no_class = SegmentSpec(room_bucket="2-к 45-60")
+        with _full_stack_patch(), patch(f"{_MOD}._query_artificial_demand", side_effect=_capture):
+            card = compute_special_indices(db, spec=spec_no_class, district=None)
+        assert captured["obj_class"] is None
+        assert captured["district"] is None
+        assert card.indices[KEY_ARTIFICIAL_DEMAND].value == pytest.approx(0.25)