gendesign/backend/app/services/forecasting/special_indices.py

"""§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   ← §25.3 ИСТИННОЕ пересечение рек. сегмента-кандидата с НАШИМ
                          портфелем (own_portfolio #1169) по 4 осям (класс/аудитория,
                          цена ₽/м², квартирография, тайминг) + гео-вес близости наших
                          проектов к участку. Портфель пуст (own_developer_ids не задан И
                          нет manual-строк) → FALLBACK на ПРОКСИ (доля конкурентов того же
                          класса, явно помеченный). DEGRADE-to-None, если нет геометрии.
  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
import math
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import date
from typing import Any, Literal

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.schemas.parcel import CompetitorsRequest
from app.services.analytics_queries import recommend_mix
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 (
    _PRICE_BAND_BUSINESS_MAX,
    _PRICE_BAND_COMFORT_MAX,
    _PRICE_BAND_ECONOMY_MAX,
    PRICE_BUCKET_BUSINESS,
    PRICE_BUCKET_COMFORT,
    PRICE_BUCKET_ECONOMY,
    PRICE_BUCKET_PREMIUM,
    SegmentSpec,
)
from app.services.forecasting.what_to_build import rank_segments
from app.services.site_finder.competitors import _CLASS_ORDER, _normalize_class, get_competitors
from app.services.site_finder.district_resolver import resolve_objective_districts
from app.services.site_finder.market_metrics import compute_market_metrics
from app.services.site_finder.own_portfolio import OwnProject, get_own_portfolio

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

# ── §25.3 own-portfolio overlap (TRUE cannibalization, #1169 PR2) ──────────────

# Похожесть аудитории/класса по дистанции в _CLASS_ORDER (competitors._CLASS_ORDER):
# тот же класс = полная каннибализация аудитории, соседний (1 шаг) — частичная,
# дальше — слабая. Чуть резче, чем competitors._CLASS_SIM_* (там relevance-
# корректировка), т.к. каннибализация бьёт ПО АУДИТОРИИ: 0 шагов→1.0, 1→0.5, 2→0.2,
# ≥3→0.05. Класс одной из сторон не распознан → ось НЕДОСТУПНА (None, НЕ 0).
_OVERLAP_CLASS_BY_STEPS: dict[int, float] = {0: 1.0, 1: 0.5, 2: 0.2}
_OVERLAP_CLASS_FAR: float = 0.05

# Период полу-затухания (мес) тайминговой близости запусков: |Δмес| = half_life →
# вклад тайминга 0.5; exp(-|Δ|/half_life). 12 мес — год между выходами ≈ половинное
# пересечение окон продаж (типовая экспозиция ЖК на рынке). Tunable.
_TIMING_HALF_LIFE_MONTHS: float = 12.0

# Масштаб (км) гео-затухания веса нашего проекта: weight = exp(-distance_km/scale).
# 3 км → вес 0.37, 1 км → 0.72, 10 км → 0.04. Ближний наш проект каннибализирует
# сильнее (общий локальный спрос); дальний — почти нет. Tunable.
_GEO_WEIGHT_SCALE_KM: float = 3.0

# Гео-вес нашего проекта без координат (тайминг/цена сравнимы, но близость неизвестна):
# НЕ исключаем проект и НЕ даём полный вес — низкий floor, чтобы он мог сигналить, но
# не доминировал над проектом с подтверждённой близостью. None-not-0 дух (есть проект,
# но геопривязка слабая → down-weight, а не выкидываем).
#
# ВАЖНО: значение НИЖЕ exp(−6.9/3)≈0.10 (вес подтверждённо дальнего ~6.9 км проекта),
# иначе «координаты неизвестны» даёт бо́льший вес чем «подтверждённо далеко» — инверсия.
# При scale=3 км: exp(−9/3)=exp(−3)≈0.050. Значение 0.05 ≈ «условные 9 км»: проект без
# координат не перебивает подтверждённо дальний (>6.9 км), но сохраняет ненулевой сигнал.
_GEO_WEIGHT_UNKNOWN: float = 0.05

# Минимум доступных осей пересечения для пары кандидат↔наш проект, чтобы считать оценку
# надёжной. < этого (напр. только класс) → пара low-confidence (сигнал есть, но тонкий).
_OVERLAP_MIN_AXES: int = 2

# Окно (мес) истории для рекомендованной квартирографии кандидата (analytics_queries.
# recommend_mix). 24 — hard-cap самого recommend_mix (#24: данные старше нерелевантны из-за
# смены ставок/ипотеки); зеркалит дефолт recommend_mix. Tunable в пределах ≤24.
_UNIT_MIX_WINDOW_MONTHS: int = 24

# Порог «тонкого» портфеля для honest-confidence (§26): портфель только из current
# (нет future-пайплайна) ИЛИ проектов меньше этого числа → confidence='low' + нота.
_THIN_PORTFOLIO_MAX_PROJECTS: int = 2

# 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)


# ── §25.3 TRUE own-portfolio overlap — pure-оси (None-not-0) + агрегация ────────


def _class_overlap(candidate_class: str | None, own_class: str | None) -> float | None:
    """Ось аудитория/класс: пересечение классов кандидата и нашего проекта. PURE.

    Нормализуем оба класса (_normalize_class — регистро/язык-независимо: 'Комфорт'/
    'comfort'/'комфорт-класс' → один ключ) и берём похожесть по дистанции в _CLASS_ORDER:
    0 шагов (тот же класс) → 1.0, 1 (соседний) → 0.5, 2 → 0.2, ≥3 → 0.05. Класс ХОТЯ БЫ
    одной стороны не распознан → None (ось НЕДОСТУПНА — НЕ фабрикуем 0). PURE.
    """
    c = _normalize_class(candidate_class)
    o = _normalize_class(own_class)
    if c is None or o is None:
        return None
    steps = abs(_CLASS_ORDER[c] - _CLASS_ORDER[o])
    return _OVERLAP_CLASS_BY_STEPS.get(steps, _OVERLAP_CLASS_FAR)


def _price_bucket_to_band(price_bucket: str | None) -> tuple[float, float] | None:
    """price_bucket кандидата → ценовая вилка [lo, hi] ₽/м² (зеркало _PRICE_BAND_*). PURE.

    Кандидат-сегмент несёт лишь bucket-ключ (не вилку); разворачиваем его в тот же
    диапазон, что sales_series.price_bucket_of: эконом [0, 120k), комфорт [120k, 160k),
    бизнес [160k, 220k), премиум [220k, +inf). Неизвестный/None bucket → None (ось цены
    НЕДОСТУПНА — НЕ фабрикуем). PURE.
    """
    if price_bucket == PRICE_BUCKET_ECONOMY:
        return (0.0, _PRICE_BAND_ECONOMY_MAX)
    if price_bucket == PRICE_BUCKET_COMFORT:
        return (_PRICE_BAND_ECONOMY_MAX, _PRICE_BAND_COMFORT_MAX)
    if price_bucket == PRICE_BUCKET_BUSINESS:
        return (_PRICE_BAND_COMFORT_MAX, _PRICE_BAND_BUSINESS_MAX)
    if price_bucket == PRICE_BUCKET_PREMIUM:
        return (_PRICE_BAND_BUSINESS_MAX, math.inf)
    return None


def _price_overlap(
    candidate_band: tuple[float, float] | None,
    own_lo: float | None,
    own_hi: float | None,
) -> float | None:
    """Ось цены: доля пересечения ценовых вилок кандидата и нашего проекта. PURE.

    Пересечение интервалов [c_lo,c_hi] и [o_lo,o_hi]; нормируем на УЖЕ узкую вилку
    (min ширины) → полное накрытие более узкого диапазона = 1.0 (стороны полностью
    конкурируют в этой ценовой зоне). Любая граница / вилка кандидата неизвестна →
    None (ось НЕДОСТУПНА — НЕ фабрикуем 0). Открытая правая граница (премиум, +inf)
    поддержана: ширина считается от конечной нижней.

    Вырожденная вилка (point: own_min==own_max или c_lo==c_hi) обрабатывается ДО
    нормирования: точка ВНУТРИ другой вилки → 1.0 (полное накрытие узкого = вырожденная
    вилка нулевой ширины полностью «попадает» в широкую — предельный случай 1.0),
    точка ВНЕ → 0.0. Обе вилки — точки: совпадают → 1.0, различны → 0.0. Это устраняет
    разрыв [148k,152k]→1.0 vs [150k,150k]→0.0 (см. #1224). PURE.
    """
    if candidate_band is None or own_lo is None or own_hi is None:
        return None
    c_lo, c_hi = candidate_band
    o_lo, o_hi = (own_lo, own_hi) if own_lo <= own_hi else (own_hi, own_lo)
    lo = max(c_lo, o_lo)
    hi = min(c_hi, o_hi)
    # Обе вилки уходят в +inf (оба премиум) — считаем полным пересечением.
    # Проверяется ДО арифметики ширин (inf - inf = nan).
    if math.isinf(c_hi) and math.isinf(o_hi):
        return 1.0
    c_width = c_hi - c_lo
    o_width = o_hi - o_lo
    # Вырожденный случай (≥1 вилка — точка): фильтр w>0 ниже выкинул бы нулевую ширину
    # и дал бы 0/positive=0.0 даже когда точка ЛЕЖИТ внутри другой вилки. Это разрыв
    # ([148k,152k]→1.0 vs [150k,150k]→0.0) и противоречит докстрингу «накрытие узкого =
    # 1.0». Обрабатываем геометрией включения: точка внутри [lo,hi] → 1.0, вне → 0.0.
    if c_width == 0 or o_width == 0:
        return 1.0 if lo <= hi else 0.0
    overlap = max(0.0, hi - lo)
    return _clamp01(overlap / min(c_width, o_width))


# Канонические ключи квартирографии (комнатность) — общее пространство, в котором
# СРАВНИВАЮТСЯ доли кандидата и нашего проекта. Кандидатный микс приходит из
# analytics_queries.recommend_mix (ключи = _BUCKET_PRETTY: «Студии 15-30»/«1-к 30-45»/
# …, зеркало sales_series.ROOM_AREA_BUCKET_*), наш — из own_planned_project.unit_mix
# (manual JSONB, конвенция миграции 148: {"studio","1k","2k","3k"}). Словари РАЗНЫЕ →
# без приведения к одному ключу L1-похожесть всегда дала бы 0 (нет общих ключей). Поэтому
# обе стороны фолдим в эти 5 канонов по комнатности (площадь-bucket «80+ м²» ≈ 4к+).
_CANON_BUCKET_STUDIO: str = "studio"
_CANON_BUCKET_1K: str = "1k"
_CANON_BUCKET_2K: str = "2k"
_CANON_BUCKET_3K: str = "3k"
_CANON_BUCKET_4K_PLUS: str = "4k+"

# Явный фолдинг сырых ключей (нижний регистр, без пробелов по краям) → канон. Покрыты обе
# конвенции: recommend_mix/_BUCKET_PRETTY (RU-подписи с диапазоном площади) И manual
# own_planned_project (латиница/частые RU-варианты). Неузнанный ключ НЕ фолдится — остаётся
# как есть (см. _canonical_room_bucket): неизвестные форматы всё равно само-сопоставляются,
# а не теряются. Детерминированно (статичная таблица).
_ROOM_BUCKET_CANON: dict[str, str] = {
    # recommend_mix / _BUCKET_PRETTY (= sales_series.ROOM_AREA_BUCKET_*)
    "студии 15-30": _CANON_BUCKET_STUDIO,
    "1-к 30-45": _CANON_BUCKET_1K,
    "2-к 45-60": _CANON_BUCKET_2K,
    "3-к 60-80": _CANON_BUCKET_3K,
    "80+ м²": _CANON_BUCKET_4K_PLUS,
    # manual own_planned_project (миграция 148): латиница
    "studio": _CANON_BUCKET_STUDIO,
    "1k": _CANON_BUCKET_1K,
    "2k": _CANON_BUCKET_2K,
    "3k": _CANON_BUCKET_3K,
    "4k": _CANON_BUCKET_4K_PLUS,
    "4k+": _CANON_BUCKET_4K_PLUS,
    "5k+": _CANON_BUCKET_4K_PLUS,
    # частые RU-варианты ручного ввода (короткие)
    "студия": _CANON_BUCKET_STUDIO,
    "студии": _CANON_BUCKET_STUDIO,
    "1к": _CANON_BUCKET_1K,
    "2к": _CANON_BUCKET_2K,
    "3к": _CANON_BUCKET_3K,
    "4к": _CANON_BUCKET_4K_PLUS,
    "4к+": _CANON_BUCKET_4K_PLUS,
}


def _canonical_room_bucket(label: str) -> str:
    """Привести сырой ключ квартирографии к канону комнатности. PURE.

    Регистро/пробел-независимо фолдит и RU-подписи recommend_mix («1-к 30-45»), и
    латиницу/короткие RU-варианты ручного own_planned_project («1k»/«1к») в один канон
    (studio/1k/2k/3k/4k+), чтобы доли кандидата и нашего проекта сравнивались в ОБЩЕМ
    пространстве. Неузнанный формат → нормализованный (lower+strip) ключ как есть: такой
    bucket само-сопоставится (одинаковая метка с обеих сторон), но не «склеится» с чужим.
    PURE.
    """
    norm = label.strip().lower()
    return _ROOM_BUCKET_CANON.get(norm, norm)


def _unit_mix_similarity(
    candidate_mix: dict[str, float] | None,
    own_mix: dict[str, float] | None,
) -> float | None:
    """Ось квартирографии: похожесть распределений долей. PURE.

    similarity = 1 − 0.5·Σ|p_k − q_k| по ОБЪЕДИНЕНИЮ ключей (отсутствующий ключ = доля 0).
    Для нормированных распределений (Σ=1) это ∈ [0,1]: идентичные миксы → 1.0,
    непересекающиеся → 0.0. Доли нормируем на свою сумму (страховка от «не-в-сумме-1»
    входа) И приводим ключи к канону комнатности (_normalize_shares), чтобы RU-словарь
    recommend_mix и латиница manual-проекта сравнивались в общем пространстве. Любой микс
    None/пустой → None (ось НЕДОСТУПНА — НЕ фабрикуем). PURE, не мутирует вход.
    """
    if not candidate_mix or not own_mix:
        return None
    cand = _normalize_shares(candidate_mix)
    own = _normalize_shares(own_mix)
    if cand is None or own is None:
        return None
    keys = set(cand) | set(own)
    l1 = sum(abs(cand.get(k, 0.0) - own.get(k, 0.0)) for k in sorted(keys))
    return _clamp01(1.0 - 0.5 * l1)


def _normalize_shares(mix: dict[str, float]) -> dict[str, float] | None:
    """Нормировать доли на сумму → {канон-ключ: share}, Σ=1; ключи к канону комнатности. PURE.

    Отрицательные/None доли отбрасываем (грязь). Ключи приводим к канону
    (_canonical_room_bucket) и СУММИРУЕМ доли, схлопнувшиеся в один канон (страховка от
    дублей вроде «studio»+«студия»). Сумма ≤ 0 → None (нет осмысленного распределения).
    Ключи сортируются для детерминизма обхода. PURE, не мутирует вход.
    """
    clean: dict[str, float] = {}
    for k, v in mix.items():
        if v is None or float(v) <= 0:
            continue
        clean[_canonical_room_bucket(k)] = clean.get(_canonical_room_bucket(k), 0.0) + float(v)
    total = sum(clean.values())
    if total <= 0:
        return None
    return {k: clean[k] / total for k in sorted(clean)}


def _timing_overlap(
    candidate_month: date | None,
    own_month: date | None,
    *,
    half_life_months: float = _TIMING_HALF_LIFE_MONTHS,
) -> float | None:
    """Ось тайминга: временна́я близость окон запуска. PURE.

    exp(−|Δмесяцев| / half_life): одновременный выход → 1.0, расхождение в half_life мес
    → 0.5, дальше затухает. Чем ближе наши запуски, тем сильнее пересекаются окна продаж
    = выше каннибализация. Любая дата None → None (ось НЕДОСТУПНА — НЕ фабрикуем). PURE.
    """
    if candidate_month is None or own_month is None:
        return None
    months_apart = abs(_months_between(candidate_month, own_month))
    if half_life_months <= 0:
        return 1.0 if months_apart == 0 else 0.0
    # Истинный период полу-затухания: 0.5 ** (Δ/half_life) → ровно 0.5 при Δ=half_life.
    return _clamp01(0.5 ** (months_apart / half_life_months))


def _months_between(a: date, b: date) -> int:
    """Целое число месяцев между двумя датами (знак = направление). PURE."""
    return (a.year - b.year) * 12 + (a.month - b.month)


def _add_months(base: date, months: int) -> date:
    """Прибавить N месяцев к дате → ПЕРВОЕ число итогового месяца. PURE.

    Без сторонних зависимостей (НЕ dateutil): год-границу считаем вручную через divmod
    (зеркалит macro_series._shift_months). months может быть отрицательным/нулём. Результат
    нормализован к 1-му числу — совпадает с тем, как own-портфель хранит release_month
    (own_portfolio._month_start) и как тайминговая ось сравнивает запуски. Напр. октябрь +
    6 → апрель следующего года. Детерминированно.
    """
    total = base.year * 12 + (base.month - 1) + months
    year, month0 = divmod(total, 12)
    return date(year, month0 + 1, 1)


def _geo_weight(
    distance_km: float | None,
    *,
    scale_km: float = _GEO_WEIGHT_SCALE_KM,
    unknown_weight: float = _GEO_WEIGHT_UNKNOWN,
) -> float:
    """Гео-вес нашего проекта по близости к участку кандидата. PURE.

    exp(−distance_km / scale): на участке (0 км) → 1.0, дальше затухает (1 км ≈ 0.72,
    3 км ≈ 0.37). Расстояние неизвестно (нет координат у проекта/участка) → НИЗКИЙ floor
    `unknown_weight` (проект ещё может сигналить, но не доминирует — близость не
    подтверждена). Это ВЕС (не ось пересечения), поэтому всегда задан (не None). PURE.

    Args:
        distance_km: расстояние от участка до нашего проекта (км) или None.
        scale_km: масштаб затухания.
        unknown_weight: floor-вес при неизвестном расстоянии.

    Returns:
        Вес ∈ (0,1].
    """
    if distance_km is None:
        return unknown_weight
    if scale_km <= 0:
        return 1.0 if distance_km <= 0 else 0.0
    return _clamp01(math.exp(-max(0.0, distance_km) / scale_km))


@dataclass(frozen=True)
class _PairOverlap:
    """Пересечение кандидата с ОДНИМ нашим проектом (промежуточный, не публичный).

    `overlap` — среднее ДОСТУПНЫХ осей ∈ [0,1] или None (ни одной оси). `n_axes` —
    сколько осей реально посчитано (None-оси исключены). `geo_weight` — вес близости.
    `signal` = overlap × geo_weight (гео-взвешенный вклад проекта). `axes` — пер-ось
    breakdown для explainability.
    """

    overlap: float | None
    n_axes: int
    geo_weight: float
    signal: float | None
    axes: dict[str, float | None]


def _own_portfolio_overlap(
    *,
    class_overlap: float | None,
    price_overlap: float | None,
    unit_mix_overlap: float | None,
    timing_overlap: float | None,
    geo_weight: float,
) -> _PairOverlap:
    """Свести оси пересечения кандидат↔наш-проект в гео-взвешенный сигнал. PURE.

    Усредняем ТОЛЬКО доступные (не-None) оси — пропущенная ось НЕ тянет пересечение к 0
    (None-not-0): нет данных по оси → её просто нет в среднем. Ни одной оси → overlap
    None (пара неинформативна). signal = overlap × geo_weight: близкий и сильно
    пересекающийся наш проект даёт высокий сигнал каннибализации. PURE.

    Args:
        class_overlap/price_overlap/unit_mix_overlap/timing_overlap: оси ∈ [0,1] | None.
        geo_weight: вес близости проекта к участку ∈ (0,1].

    Returns:
        _PairOverlap (overlap=None, signal=None, если ни одной оси не доступно).
    """
    axes: dict[str, float | None] = {
        "class": class_overlap,
        "price": price_overlap,
        "unit_mix": unit_mix_overlap,
        "timing": timing_overlap,
    }
    available = [v for v in axes.values() if v is not None]
    if not available:
        return _PairOverlap(overlap=None, n_axes=0, geo_weight=geo_weight, signal=None, axes=axes)
    overlap = sum(available) / len(available)
    return _PairOverlap(
        overlap=overlap,
        n_axes=len(available),
        geo_weight=geo_weight,
        signal=overlap * geo_weight,
        axes=axes,
    )


def _aggregate_overlap(pairs: Sequence[_PairOverlap]) -> float | None:
    """Свести пер-проектные гео-взвешенные сигналы в индекс каннибализации. PURE.

    Берём МАКСИМУМ signal (= overlap × geo_weight) по нашим проектам, а НЕ среднее:
    ОДИН близкий сильно-пересекающийся проект — уже самостоятельный риск каннибализации,
    и среднее размыло бы его длинным хвостом далёких/непохожих проектов (soft-max=max —
    «сильнейший каннибализатор»). Ни одной пары с сигналом → None (нечего агрегировать,
    НЕ 0). Детерминированно (max по числам). PURE.
    """
    signals = [p.signal for p in pairs if p.signal is not None]
    if not signals:
        return None
    return _clamp01(max(signals))


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 — все опциональны.
# District-фильтр: admin-имя (вход в caller) РАЗВОРАЧИВАЕТСЯ через
# `resolve_objective_districts` в набор МИКРО-районов (`ol.district` хранит
# микро-вокабуляр: 'Уралмаш', 'ЖБИ', ...; admin 'Кировский' там НЕ встречается →
# сырая фильтрация по admin-имени схлопывалась в 0 строк — #1205, тот же класс
# что #1211 в _price_sensitivity). Зеркало `sales_series._SOURCE_B_SQL` /
# `market_metrics._SALES_WINDOW_SQL`: bind `:has_district` (boolean) переключает
# ветку, `:districts` (text[]) — список микро. «Продан» = is_sold IS TRUE ИЛИ
# contract_date заполнена. 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(:has_district AS boolean) IS FALSE
                OR ol.district = ANY(CAST(:districts 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). Детерминированно.

    District-фильтр: admin-имя ЕКБ → набор МИКРО-районов через
    `resolve_objective_districts` (admin не встречается в `objective_lots.district` —
    там микро-вокабуляр). None / unresolved → EKB-wide (без district-фильтра).
    Лечит #1205 (silent-correctness: admin 'Кировский' → 0 строк → ложное
    'unavailable' в district-scoped отчётах).
    """
    micros = resolve_objective_districts(db, district)
    row = (
        db.execute(
            _ARTIFICIAL_DEMAND_SQL,
            {
                "premise_kind": premise_kind,
                # has_district=False → district-ветка фильтра отключена (EKB-wide).
                "has_district": micros is not None,
                # ANY(NULL::text[]) валиден; пустой список когда фильтра нет.
                "districts": micros if micros is not None else [],
                "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),
    }


# ──────────────────────────────────────────────────────────────────────────────
# §25.3 cannibalization — центроид участка (гео-вес наших проектов). psycopg v3.
# ──────────────────────────────────────────────────────────────────────────────

# Центроид участка кандидата (lon/lat) для гео-веса наших проектов — reuse семантики
# competitors._PARCEL_CENTROID_SQL (cad_parcels_geom → fallback cad_quarters_geom).
# Параметризован, без f-string. Гео тут — ВЕС (не ось пересечения): нет геометрии →
# все гео-веса падают на _GEO_WEIGHT_UNKNOWN (overlap всё равно считается по остальным
# осям). НИКОГДА :x::type.
_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
    """
)

# Средний радиус Земли (км) для гаверсинуса. Распределённость наших проектов в пределах
# города — гаверсинус точен достаточно (PostGIS-geography для пары точек тут избыточен).
_EARTH_RADIUS_KM: float = 6371.0


def _quarter_from_cad(cad_num: str) -> str:
    """Кадастровый квартал из номера участка (66:41:0303161:123 → 66:41:0303161). PURE.

    Нестандартный формат → cad_num как есть (fallback). Зеркалит competitors._quarter_from_cad.
    """
    parts = cad_num.split(":")
    if len(parts) >= 3:
        return ":".join(parts[:3])
    return cad_num


def _haversine_km(lon1: float, lat1: float, lon2: float, lat2: float) -> float:
    """Гаверсинус-расстояние (км) между двумя (lon,lat). PURE.

    Детерминированно; для внутригородских пар точек точности достаточно (гео — лишь вес).
    """
    rlat1, rlat2 = math.radians(lat1), math.radians(lat2)
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    h = math.sin(dlat / 2) ** 2 + math.cos(rlat1) * math.cos(rlat2) * math.sin(dlon / 2) ** 2
    return 2.0 * _EARTH_RADIUS_KM * math.asin(min(1.0, math.sqrt(h)))


def _query_parcel_centroid(db: Session, *, cad_num: str) -> tuple[float, float] | None:
    """Центроид участка (lon, lat) для гео-веса. Graceful → None.

    Нет геометрии / сбой → None (гео-веса упадут на floor; overlap считается по остальным
    осям, индекс НЕ деградирует целиком). Параметризовано (psycopg v3). Детерминированно.
    """
    try:
        row = (
            db.execute(
                _PARCEL_CENTROID_SQL,
                {"cad_num": cad_num, "quarter": _quarter_from_cad(cad_num)},
            )
            .mappings()
            .first()
        )
    except Exception:
        logger.exception("cannibalization: centroid query failed for cad_num=%s", cad_num)
        return None
    if not row or row["lon"] is None or row["lat"] is None:
        return None
    return (float(row["lon"]), float(row["lat"]))


# ──────────────────────────────────────────────────────────────────────────────
# 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 _launch_window_horizon(launch_window: SpecialIndex) -> int | None:
    """Извлечь рекомендованный горизонт запуска (мес) из готового Launch Window. PURE.

    §25.1 кладёт выбранный горизонт пикового дефицита в detail['best_horizon_months'].
    Индекс недоступен (deficit None на всех горизонтах → method='unavailable', detail без
    горизонта) → None: тайминговая ось §25.3 НЕ активируется (graceful, НЕ фабрикуем дату).
    Читаем из УЖЕ посчитанного индекса — Launch Window НЕ пересчитывается. PURE.
    """
    horizon = launch_window.detail.get("best_horizon_months")
    return horizon if isinstance(horizon, int) else None


def _candidate_release_month(launch_window: SpecialIndex, *, as_of: date) -> date | None:
    """Когда рекомендованный проект реально выйдет на рынок (тайминг §25.3). PURE.

    = as_of (дата отчёта) + горизонт окна запуска §25.1: на этот месяц #980 предсказывает
    ПИКОВЫЙ дефицит, т.е. рекомендованный к запуску момент. Этот месяц затем сравнивается с
    release_month наших проектов (тайминговая ось каннибализации). Launch Window недоступен
    → None: ось тайминга остаётся исключённой (None-not-0), а не фабрикуется. PURE.
    """
    horizon = _launch_window_horizon(launch_window)
    if horizon is None:
        return None
    return _add_months(as_of, horizon)


def _candidate_unit_mix_from_recommend(mix_result: Any) -> dict[str, float] | None:
    """Извлечь распределение долей {bucket: share} из ответа recommend_mix. PURE.

    recommend_mix (analytics_queries) кладёт рекомендованную квартирографию в
    `buckets` — список {`bucket`: подпись (= _BUCKET_PRETTY / sales_series.
    ROOM_AREA_BUCKET_*), `share_pct`: доля сегмента в %, Σ≈100, …}. Берём пары
    (bucket, share_pct>0) — это и есть SHARE-распределение. Ключи остаются в RU-
    словаре recommend_mix; к канону комнатности их приведёт _normalize_shares внутри
    оси похожести (как и ключи нашего проекта), поэтому здесь НЕ нормируем и НЕ
    переименовываем. `buckets` пуст/нет осмысленных долей (или вход не dict — defensive,
    recommend_mix loosely-typed dict[str, Any]) → None (ось НЕДОСТУПНА — НЕ фабрикуем).
    `mix_result: Any` — парсим разнородную dict-форму recommend_mix защитно. PURE.
    """
    if not isinstance(mix_result, dict):
        return None
    buckets = mix_result.get("buckets")
    if not isinstance(buckets, list):
        return None
    mix: dict[str, float] = {}
    for b in buckets:
        if not isinstance(b, dict):
            continue
        bucket = b.get("bucket")
        share = b.get("share_pct")
        if isinstance(bucket, str) and isinstance(share, int | float) and share > 0:
            mix[bucket] = mix.get(bucket, 0.0) + float(share)
    return mix or None


def _build_candidate_unit_mix(
    db: Session,
    *,
    spec: SegmentSpec,
    district: str | None,
    cad_num: str | None,
) -> dict[str, float] | None:
    """Рекомендованная квартирография кандидата для оси §25.3. Graceful → None.

    Зовёт analytics_queries.recommend_mix (та же rule-based квартирография, что в §22
    product_tz) и достаёт SHARE-распределение (_candidate_unit_mix_from_recommend).
    Требует район (recommend_mix района-центрик: bucket-распределение город-wide, но вход
    обязателен) — нет district → None (ось исключается, как при недоступном Launch Window
    у тайминга). Любой сбой/пустой ответ → None (НЕ crash, НЕ фабрикуем микс): ось
    квартирографии просто исключается из среднего, каннибализация считается по остальным.

    COST (#1129): recommend_mix — тяжёлый запрос и в report-пути НЕ вызывается отдельно
    (overlay §9.7 — иная функция), поэтому это доп. вызов на отчёт. Его дорогие
    под-загрузки (build_sales_series / compute_market_metrics / get_competitors) обёрнуты
    @cached и внутри forecast_cache() становятся попаданиями; некэшируемый остаток —
    SQL самого recommend_mix (bucket-распределение/velocity/elasticity/comparables).
    horizon_months НЕ передаём — overlay §9.7 здесь не нужен (нам только живой mix).
    """
    if not district:
        return None
    try:
        result = recommend_mix(
            db,
            district_name=district,
            target_class=spec.obj_class,
            months_window=_UNIT_MIX_WINDOW_MONTHS,
            cad_num=cad_num,
        )
    except Exception:
        logger.exception(
            "cannibalization: recommend_mix failed (district=%s cad_num=%s) "
            "→ ось квартирографии исключена (None-not-0)",
            district,
            cad_num,
        )
        return None
    return _candidate_unit_mix_from_recommend(result)


def _portfolio_has_unit_mix(portfolio: Sequence[OwnProject]) -> bool:
    """Есть ли в портфеле ХОТЯ БЫ один проект с непустой квартирографией. PURE.

    Гейт hot-path (#1129): ось квартирографии §25.3 может внести вклад ТОЛЬКО если у
    нашего проекта задан unit_mix (current/domrf-проекты несут unit_mix=None — см.
    own_portfolio PR1; микс есть лишь у manual-future). Нет ни одного проекта с миксом →
    кандидатный recommend_mix (ТЯЖЁЛЫЙ запрос) считать НЕЗАЧЕМ: ось всё равно
    исключится (None-not-0). Пустой/None-микс/пустой dict не считаются «миксом». PURE.
    """
    return any(p.unit_mix for p in portfolio)


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,
    )


# Метка-каведат для проксированной каннибализации (own-portfolio не сконфигурирован).
_PROXY_REASON: str = (
    "own-portfolio не сконфигурирован (own_developer_ids не задан И нет manual-строк) "
    "→ ПРОКСИ: плотность конкурентов того же класса (НЕ истинное пересечение портфеля)"
)


def _build_cannibalization_proxy(
    db: Session, *, spec: SegmentSpec, cad_num: str | None
) -> SpecialIndex:
    """Index 3 FALLBACK — ПРОКСИ: доля relevance-силы конкурентов В классе. Degrade.

    Используется ТОЛЬКО когда наш портфель пуст (own-portfolio не сконфигурирован):
    честно помечается method='proxy_*' и detail.proxy=True, чтобы прокси НИКОГДА не
    выдавался за истинный own-portfolio индекс. Логика — прежний _cannibalization_index.
    """
    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 = _normalize_class(spec.obj_class)
    same_class_relevance = [
        c.relevance_weight
        for c in competitors
        if target is not None and _normalize_class(c.obj_class) == target
    ]
    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=f"прокси · {spec.obj_class}" if spec.obj_class else "прокси",
        confidence=_cap_confidence("low"),  # прокси — заведомо слабее истинного
        detail={
            "mode": "proxy",
            "proxy": True,
            "proxy_reason": _PROXY_REASON,
            "n_competitors": len(competitors),
            "n_same_class": len(same_class_relevance),
            "target_class": spec.obj_class,
            "radius_km": _COMPETITOR_RADIUS_KM,
        },
        method="proxy_same_class_relevance_share",
        advisory=_ADVISORY,
    )


def _build_cannibalization_true(
    *,
    spec: SegmentSpec,
    portfolio: Sequence[OwnProject],
    centroid: tuple[float, float] | None,
    candidate_unit_mix: dict[str, float] | None,
    candidate_release_month: date | None,
) -> SpecialIndex:
    """Index 3 TRUE — §25.3 пересечение кандидата с НАШИМ портфелем по 4 осям + гео.

    Для каждого нашего проекта считаем оси (класс/цена/квартирография/тайминг —
    None-not-0 при тонких данных) + гео-вес близости к участку, сводим в гео-взвешенный
    сигнал, агрегируем МАКСИМУМОМ (сильнейший каннибализатор). Топ-проекты — в detail
    (explainability §16). Confidence отражает покрытие портфеля (тонкий/только-current →
    low + нота §26). Чистая агрегация над pure-осями — без БД здесь. Детерминированно.
    """
    candidate_band = _price_bucket_to_band(spec.price_bucket)
    pairs: list[_PairOverlap] = []
    contributions: list[tuple[float, OwnProject, _PairOverlap]] = []
    for own in portfolio:
        distance_km = _own_distance_km(centroid, own)
        price_overlap = _price_overlap(candidate_band, own.price_min_per_m2, own.price_max_per_m2)
        pair = _own_portfolio_overlap(
            class_overlap=_class_overlap(spec.obj_class, own.obj_class),
            price_overlap=price_overlap,
            unit_mix_overlap=_unit_mix_similarity(candidate_unit_mix, own.unit_mix),
            timing_overlap=_timing_overlap(candidate_release_month, own.release_month),
            geo_weight=_geo_weight(distance_km),
        )
        pairs.append(pair)
        if pair.signal is not None:
            contributions.append((pair.signal, own, pair))

    value = _aggregate_overlap(pairs)
    if value is None:
        return _unavailable(
            KEY_CANNIBALIZATION,
            reason=(
                "ни по одному нашему проекту нет сопоставимых осей "
                "(класс/цена/квартирография/тайминг) — пересечение неизмеримо (НЕ фабрикуем)"
            ),
        )

    # Топ-вкладчики (сильнейшие каннибализаторы) — детерминированно по signal DESC,
    # tie-break по имени проекта (стабильно, без RNG/порядка set).
    contributions.sort(key=lambda t: (-t[0], t[1].name))
    top = contributions[:_COMPETITOR_TOP_N]
    confidence = _cap_confidence(_portfolio_confidence(portfolio))
    return SpecialIndex(
        key=KEY_CANNIBALIZATION,
        value=value,
        label=_top_contributor_label(top),
        confidence=confidence,
        detail={
            "mode": "own_portfolio",
            "proxy": False,
            "n_own_projects": len(portfolio),
            "n_current": sum(1 for p in portfolio if p.source == "current"),
            "n_future": sum(1 for p in portfolio if p.source == "future"),
            "has_geometry": centroid is not None,
            "axes_available": _axes_available_summary(pairs),
            "top_contributors": [_contribution_detail(s, own, pair) for s, own, pair in top],
            "confidence_note": _portfolio_confidence_note(portfolio),
        },
        method="own_portfolio_overlap",
        advisory=_ADVISORY,
    )


def _own_distance_km(centroid: tuple[float, float] | None, own: OwnProject) -> float | None:
    """Расстояние (км) от центроида участка до нашего проекта или None. PURE.

    Нет центроида ИЛИ нет координат проекта → None (гео-вес упадёт на floor). PURE.
    """
    if centroid is None or own.lon is None or own.lat is None:
        return None
    return _haversine_km(centroid[0], centroid[1], own.lon, own.lat)


def _portfolio_confidence(portfolio: Sequence[OwnProject]) -> Confidence:
    """Honest-confidence по покрытию портфеля (§26). PURE.

    Тонкий портфель (< _THIN_PORTFOLIO_MAX_PROJECTS проектов) ИЛИ ТОЛЬКО current
    (нет future-пайплайна — мы не знаем своих будущих запусков) → 'low' (данных мало).
    Иначе 'medium' (всё равно ≤ cap: слой advisory). PURE.
    """
    has_future = any(p.source == "future" for p in portfolio)
    if len(portfolio) < _THIN_PORTFOLIO_MAX_PROJECTS or not has_future:
        return "low"
    return "medium"


def _portfolio_confidence_note(portfolio: Sequence[OwnProject]) -> str | None:
    """RU-нота §26 о тонком покрытии портфеля или None, если покрытие достаточное. PURE."""
    n = len(portfolio)
    has_future = any(p.source == "future" for p in portfolio)
    if n < _THIN_PORTFOLIO_MAX_PROJECTS:
        return f"данных мало: в портфеле {n} проект(ов) — оценка ориентировочная"
    if not has_future:
        return "только текущие проекты (нет future-пайплайна) — будущие запуски не учтены"
    return None


def _axes_available_summary(pairs: Sequence[_PairOverlap]) -> dict[str, int]:
    """Сколько пар имеют каждую ось доступной (explainability «что реально считалось»). PURE."""
    summary = {"class": 0, "price": 0, "unit_mix": 0, "timing": 0}
    for pair in pairs:
        for axis, val in pair.axes.items():
            if val is not None:
                summary[axis] += 1
    return summary


def _contribution_detail(signal: float, own: OwnProject, pair: _PairOverlap) -> dict[str, Any]:
    """Explainability-карточка одного топ-каннибализатора. PURE."""
    return {
        "name": own.name,
        "source": own.source,
        "obj_class": own.obj_class,
        "district": own.district,
        "signal": _round_or_none(signal, 3),
        "overlap": _round_or_none(pair.overlap, 3),
        "geo_weight": _round_or_none(pair.geo_weight, 3),
        "n_axes": pair.n_axes,
        "axes": {k: _round_or_none(v, 3) for k, v in pair.axes.items()},
        "low_confidence_pair": pair.n_axes < _OVERLAP_MIN_AXES,
    }


def _top_contributor_label(
    top: Sequence[tuple[float, OwnProject, _PairOverlap]],
) -> str | None:
    """Человекочитаемый label = сильнейший наш каннибализатор. PURE."""
    if not top:
        return None
    _, own, _ = top[0]
    return f"пересечение с «{own.name}»"


def _build_cannibalization(
    db: Session,
    *,
    spec: SegmentSpec,
    cad_num: str | None,
    candidate_unit_mix: dict[str, float] | None = None,
    candidate_release_month: date | None = None,
    own_projects: list[OwnProject] | None = None,
) -> SpecialIndex:
    """Index 3 — Cannibalization (§25.3): истинное own-portfolio-пересечение / прокси.

    Портфель НЕпуст → истинный индекс (overlap кандидата с нашими проектами по 4 осям +
    гео-вес близости к участку); explainability называет режим own-portfolio и топ-
    каннибализаторов. Портфель ПУСТ (own_developer_ids не задан И нет manual-строк) →
    FALLBACK на ПРОКСИ (_build_cannibalization_proxy), явно помеченный method='proxy_*'
    — прокси НИКОГДА не выдаётся за истинный индекс. Детерминированно.

    candidate_release_month — месяц выхода кандидата на рынок; оркестратор выводит его
    из §25.1 Launch Window (дата отчёта + горизонт пикового дефицита) и передаёт сюда,
    активируя тайминговую ось. Launch Window недоступен → None → ось тайминга честно
    ИСКЛЮЧАЕТСЯ (None-not-0, НЕ фабрикуется). candidate_unit_mix — рекомендованный микс
    кандидата ({bucket: доля}); оркестратор выводит его из recommend_mix и передаёт сюда,
    активируя ось квартирографии. recommend_mix недоступен/нет района → None → ось
    квартирографии честно ИСКЛЮЧАЕТСЯ (None-not-0). Любая не переданная ось ИСКЛЮЧАЕТСЯ
    из среднего, а НЕ зануляется.

    own_projects — УЖЕ собранный портфель (оркестратор фетчит get_own_portfolio ОДИН раз
    и прокидывает сюда, чтобы не дёргать БД дважды за отчёт). None (дефолт) → фетчим
    внутри сами (backward-compat для прямых вызовов/тестов). Список передан → реюзаем его.

    ЧЕСТНОСТЬ (own_portfolio PR1): «текущие» (domrf) наши проекты несут unit_mix=None —
    ось квартирографии РЕАЛЬНО вносит вклад только для manual-future проектов с заданным
    миксом. Узость ожидаема (ось корректна, но разреживается до заведения портфеля).
    """
    portfolio = own_projects if own_projects is not None else get_own_portfolio(db)
    if not portfolio:
        logger.info(
            "cannibalization: own-portfolio пуст → FALLBACK на прокси (cad_num=%s)", cad_num
        )
        return _build_cannibalization_proxy(db, spec=spec, cad_num=cad_num)

    # Гео — лишь ВЕС: нет геометрии → веса на floor, overlap считается по остальным осям.
    centroid = _query_parcel_centroid(db, cad_num=cad_num) if cad_num else None
    return _build_cannibalization_true(
        spec=spec,
        portfolio=portfolio,
        centroid=centroid,
        candidate_unit_mix=candidate_unit_mix,
        candidate_release_month=candidate_release_month,
    )


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.

    §25.1 Launch Window считается ПЕРВЫМ и переиспользуется: его горизонт пикового
    дефицита + дата отчёта (date.today() — даты отчёта в scope нет) дают месяц выхода
    кандидата на рынок, который передаётся в §25.3 каннибализацию и активирует её
    тайминговую ось. Launch Window недоступен → месяц None → ось тайминга честно
    исключается (None-not-0), каннибализация считается по классу/цене/гео.

    §25.3 own-portfolio фетчится ОДИН раз и реюзается (gate квартирографии + сама
    каннибализация — никакого двойного запроса). Ось квартирографии активируется из
    recommend_mix (рекомендованный микс кандидата, та же rule-based квартирография §22
    product_tz), но recommend_mix — ТЯЖЁЛЫЙ запрос (#1129 держит §22-отчёт ~12.6s),
    поэтому зовём его ТОЛЬКО когда в портфеле есть проект с unit_mix (current/domrf несут
    None — ось всё равно исключилась бы, см. own_portfolio PR1): иначе тяжёлый вызов
    отработал бы вхолостую. Микс выводится ОДИН раз и передаётся в каннибализацию рядом с
    месяцем тайминга. recommend_mix недоступен/нет района/нет проекта с миксом → микс None
    → ось квартирографии честно исключается (None-not-0). Так §25.3 задействует все 4 оси
    (класс + цена + тайминг + квартирография) + гео-вес.

    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()

    def _run(key: str, builder: Any) -> SpecialIndex:
        """Выполнить builder одного индекса в собственном try/except (graceful)."""
        try:
            return builder()  # type: ignore[no-any-return]
        except Exception:
            # Сбой одного индекса НЕ валит карточку: деградация-None, остальные считаются.
            logger.exception(
                "special_indices: index=%s failed (segment=%s district=%s cad_num=%s)",
                key,
                segment,
                district,
                cad_num,
            )
            return _unavailable(key, reason="ошибка расчёта (см. логи)")

    # §25.1 Launch Window считаем ПЕРВЫМ и ОДИН раз: §25.3 каннибализация выводит из его
    # горизонта тайминг кандидата (дата отчёта + горизонт пикового дефицита = когда проект
    # реально выйдет на рынок). Готовый индекс переиспользуется ниже (НЕ пересчитывается).
    launch_window = _run(
        KEY_LAUNCH_WINDOW,
        lambda: _build_launch_window(
            db, spec=spec, district=district, cad_num=cad_num, horizons=horizons
        ),
    )
    # report-as-of: дата отчёта в scope недоступна → date.today() (app-код — допустимо).
    # Launch Window недоступен → release_month None → тайминговая ось §25.3 исключается.
    candidate_release_month = _candidate_release_month(launch_window, as_of=date.today())

    # §25.3 own-portfolio фетчим ОДИН раз и реюзаем (gate квартирографии + сама
    # каннибализация). get_own_portfolio дёшев, когда own_developer_ids пуст (СРАЗУ [],
    # без БД — own_portfolio PR1), поэтому в типовом случае гейт почти бесплатен.
    own_projects = get_own_portfolio(db)

    # §25.3 ось квартирографии: рекомендованный микс кандидата (recommend_mix, та же
    # rule-based квартирография §22 product_tz). recommend_mix — ТЯЖЁЛЫЙ запрос (#1129
    # держит §22-отчёт ~12.6s), поэтому зовём его ТОЛЬКО когда в нашем портфеле есть хотя
    # бы один проект с unit_mix: иначе ось всё равно исключится (current/domrf несут
    # unit_mix=None) и тяжёлый recommend_mix отработал бы вхолостую. Нет такого проекта →
    # микс None (ось исключена, ровно как при None-ответе recommend_mix). Выводим ОДИН раз
    # и передаём в каннибализацию рядом с candidate_release_month. Внутри graceful (сбой/нет
    # района → None → ось исключается, None-not-0), поэтому отдельный _run не нужен.
    candidate_unit_mix = (
        _build_candidate_unit_mix(db, spec=spec, district=district, cad_num=cad_num)
        if _portfolio_has_unit_mix(own_projects)
        else None
    )

    # (ключ, нулевой-аргумент builder) — каждый вызывается в _run (общий try/except).
    builders: dict[str, Any] = {
        KEY_LAUNCH_WINDOW: lambda: launch_window,  # уже посчитан выше
        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,
            candidate_unit_mix=candidate_unit_mix,
            candidate_release_month=candidate_release_month,
            own_projects=own_projects,
        ),
        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] = {key: _run(key, builders[key]) for key in _INDEX_KEYS}

    # confidence = MIN по ДОСТУПНЫМ индексам (контракт SpecialIndices / docstring выше).
    # Недоступный индекс (_unavailable → value=None, confidence='low') НЕ участвует: его
    # 'low' — артефакт деградации, а не сигнал низкой уверенности доступных индексов
    # (#1592: _min_confidence отбрасывает только None, поэтому фильтруем здесь по value).
    confidence = _min_confidence(
        [idx.confidence for idx in indices.values() if idx.value is not None]
    )

    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,
    )