feat(forecasting): activate §25.3 cannibalization unit-mix axis (4/4), gated (#1169)

Feed candidate_unit_mix into _build_cannibalization (mirrors how #1169 fed
candidate_release_month from the launch window), completing §25.3 to all 4 axes:
class + price + timing + unit-mix, plus geo weight.

- candidate mix from recommend_mix "buckets[].share_pct" (same rule-based
  квартирография as §22 product_tz), extracted + normalized to {bucket: share}.
- _canonical_room_bucket folds recommend_mix RU labels ("1-к 30-45", "80+ м²")
  and manual own_planned_project Latin keys ("1k") into one room-count space —
  without it the L1 similarity would silently be 0 (disjoint keys).
- recommend_mix is HEAVY, so it's GATED: derived only when the own-portfolio has
  >=1 project with a non-empty unit_mix; get_own_portfolio fetched once in
  compute_special_indices and threaded into _build_cannibalization (no double
  fetch). With OWN_DEVELOPER_IDS unset (portfolio empty) → zero added cost on the
  hot §22 report path.
- Graceful (recommend_mix None/empty/raises → axis excluded, None-not-0),
  deterministic. Unit-mix only fires for manual-future own-projects with a mix
  (domrf-current carry unit_mix=None) — expected narrowness, documented.

205 tests; ruff + mypy clean. Scope: special_indices.py + test only; no deps.

Refs #1169

backend/app/services/forecasting/special_indices.py

@@ -51,6 +51,7 @@ 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 (
@@ -147,6 +148,11 @@ _GEO_WEIGHT_UNKNOWN: float = 0.1
 # надёжной. < этого (напр. только класс) → пара 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
@@ -457,6 +463,64 @@ def _price_overlap(
     return _clamp01(overlap / min(widths))
 
 
+# Канонические ключи квартирографии (комнатность) — общее пространство, в котором
+# СРАВНИВАЮТСЯ доли кандидата и нашего проекта. Кандидатный микс приходит из
+# 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,
@@ -466,8 +530,9 @@ def _unit_mix_similarity(
     similarity = 1 − 0.5·Σ|p_k − q_k| по ОБЪЕДИНЕНИЮ ключей (отсутствующий ключ = доля 0).
     Для нормированных распределений (Σ=1) это ∈ [0,1]: идентичные миксы → 1.0,
     непересекающиеся → 0.0. Доли нормируем на свою сумму (страховка от «не-в-сумме-1»
-    входа), clamp [0,1]. Любой микс None/пустой → None (ось НЕДОСТУПНА — НЕ фабрикуем).
+    входа) И приводим ключи к канону комнатности (_normalize_shares), чтобы RU-словарь
-    PURE, не мутирует вход.
+    recommend_mix и латиница manual-проекта сравнивались в общем пространстве. Любой микс
+    None/пустой → None (ось НЕДОСТУПНА — НЕ фабрикуем). PURE, не мутирует вход.
     """
     if not candidate_mix or not own_mix:
         return None
@@ -481,12 +546,18 @@ def _unit_mix_similarity(
 
 
 def _normalize_shares(mix: dict[str, float]) -> dict[str, float] | None:
-    """Нормировать доли на их сумму → {key: share}, Σ=1. PURE.
+    """Нормировать доли на сумму → {канон-ключ: share}, Σ=1; ключи к канону комнатности. PURE.
 
-    Отрицательные/None доли отбрасываем (грязь); сумма ≤ 0 → None (нет осмысленного
+    Отрицательные/None доли отбрасываем (грязь). Ключи приводим к канону
-    распределения). Ключи сортируются для детерминизма обхода. PURE.
+    (_canonical_room_bucket) и СУММИРУЕМ доли, схлопнувшиеся в один канон (страховка от
+    дублей вроде «studio»+«студия»). Сумма ≤ 0 → None (нет осмысленного распределения).
+    Ключи сортируются для детерминизма обхода. PURE, не мутирует вход.
     """
-    clean = {k: float(v) for k, v in mix.items() if v is not None and float(v) > 0}
+    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
@@ -942,6 +1013,91 @@ def _candidate_release_month(
     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,
     *,
@@ -1235,6 +1391,7 @@ def _build_cannibalization(
     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-пересечение / прокси.
 
@@ -1247,11 +1404,21 @@ def _build_cannibalization(
     candidate_release_month — месяц выхода кандидата на рынок; оркестратор выводит его
     из §25.1 Launch Window (дата отчёта + горизонт пикового дефицита) и передаёт сюда,
     активируя тайминговую ось. Launch Window недоступен → None → ось тайминга честно
-    ИСКЛЮЧАЕТСЯ (None-not-0, НЕ фабрикуется). candidate_unit_mix — опциональная ось
+    ИСКЛЮЧАЕТСЯ (None-not-0, НЕ фабрикуется). candidate_unit_mix — рекомендованный микс
-    квартирографии кандидата (phase-2: нужен project-level рекомендованный микс; пока
+    кандидата ({bucket: доля}); оркестратор выводит его из recommend_mix и передаёт сюда,
-    обычно None). Любая не переданная ось ИСКЛЮЧАЕТСЯ из среднего, а НЕ зануляется.
+    активируя ось квартирографии. recommend_mix недоступен/нет района → None → ось
+    квартирографии честно ИСКЛЮЧАЕТСЯ (None-not-0). Любая не переданная ось ИСКЛЮЧАЕТСЯ
+    из среднего, а НЕ зануляется.
+
+    own_projects — УЖЕ собранный портфель (оркестратор фетчит get_own_portfolio ОДИН раз
+    и прокидывает сюда, чтобы не дёргать БД дважды за отчёт). None (дефолт) → фетчим
+    внутри сами (backward-compat для прямых вызовов/тестов). Список передан → реюзаем его.
+
+    ЧЕСТНОСТЬ (own_portfolio PR1): «текущие» (domrf) наши проекты несут unit_mix=None —
+    ось квартирографии РЕАЛЬНО вносит вклад только для manual-future проектов с заданным
+    миксом. Узость ожидаема (ось корректна, но разреживается до заведения портфеля).
     """
-    portfolio = get_own_portfolio(db)
+    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
@@ -1394,6 +1561,17 @@ def compute_special_indices(
     тайминговую ось. 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
@@ -1441,6 +1619,25 @@ def compute_special_indices(
     # 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,  # уже посчитан выше
@@ -1448,7 +1645,12 @@ def compute_special_indices(
             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_release_month=candidate_release_month
+            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(

backend/tests/services/forecasting/test_special_indices.py

@@ -44,7 +44,9 @@ from app.services.forecasting.special_indices import (
     _artificial_demand_share,
     _avg_ticket_rub,
     _candidate_release_month,
+    _candidate_unit_mix_from_recommend,
     _cannibalization_index,
+    _canonical_room_bucket,
     _cap_confidence,
     _clamp01,
     _class_overlap,
@@ -55,9 +57,11 @@ from app.services.forecasting.special_indices import (
     _haversine_km,
     _launch_window_horizon,
     _min_confidence,
+    _normalize_shares,
     _oversupply_risk_from_deficit,
     _own_portfolio_overlap,
     _pick_launch_window,
+    _portfolio_has_unit_mix,
     _price_bucket_to_band,
     _price_overlap,
     _query_artificial_demand,
@@ -396,6 +400,133 @@ class TestUnitMixSimilarity:
         v = _unit_mix_similarity({"studio": 0.7, "1k": 0.3}, {"studio": 0.2, "1k": 0.8})
         assert v is not None and 0.0 <= v <= 1.0
 
+    def test_cross_vocabulary_alignment_recommend_vs_manual(self) -> None:
+        # КЛЮЧЕВОЕ: кандидат из recommend_mix (RU-подписи _BUCKET_PRETTY) vs наш проект
+        # (manual латиница own_planned_project). Без приведения ключей к канону L1=2 → 0;
+        # с каноном идентичные доли по комнатности → 1.0.
+        candidate = {"Студии 15-30": 0.3, "1-к 30-45": 0.4, "2-к 45-60": 0.3}
+        own = {"studio": 0.3, "1k": 0.4, "2k": 0.3}
+        assert _unit_mix_similarity(candidate, own) == pytest.approx(1.0)
+
+    def test_cross_vocabulary_partial(self) -> None:
+        # частичное совпадение через канон: студии совпали (0.5), остальное разъехалось.
+        candidate = {"Студии 15-30": 0.5, "1-к 30-45": 0.5}
+        own = {"студия": 0.5, "2к": 0.5}  # короткие RU-варианты ручного ввода
+        # canon: candidate {studio:0.5,1k:0.5}, own {studio:0.5,2k:0.5} → L1=1.0 → 0.5.
+        assert _unit_mix_similarity(candidate, own) == pytest.approx(0.5)
+
+    def test_large_bucket_folds_to_4k_plus(self) -> None:
+        # «80+ м²» (recommend_mix) ↔ «4k» (manual) оба → канон 4k+ → совпадают.
+        assert _unit_mix_similarity({"80+ м²": 1.0}, {"4k": 1.0}) == pytest.approx(1.0)
+
+
+class TestCanonicalRoomBucket:
+    def test_recommend_mix_pretty_labels(self) -> None:
+        # RU-подписи recommend_mix / _BUCKET_PRETTY → каноны комнатности.
+        assert _canonical_room_bucket("Студии 15-30") == "studio"
+        assert _canonical_room_bucket("1-к 30-45") == "1k"
+        assert _canonical_room_bucket("2-к 45-60") == "2k"
+        assert _canonical_room_bucket("3-к 60-80") == "3k"
+        assert _canonical_room_bucket("80+ м²") == "4k+"
+
+    def test_manual_latin_keys(self) -> None:
+        # латиница own_planned_project (миграция 148).
+        assert _canonical_room_bucket("studio") == "studio"
+        assert _canonical_room_bucket("1k") == "1k"
+        assert _canonical_room_bucket("4k") == "4k+"
+
+    def test_short_ru_variants(self) -> None:
+        assert _canonical_room_bucket("студия") == "studio"
+        assert _canonical_room_bucket("1к") == "1k"
+
+    def test_case_and_whitespace_insensitive(self) -> None:
+        assert _canonical_room_bucket("  STUDIO  ") == "studio"
+        assert _canonical_room_bucket("1-К 30-45") == "1k"
+
+    def test_unknown_key_self_normalized_not_dropped(self) -> None:
+        # неузнанный формат → нормализованный (lower+strip) ключ как есть (само-сопоставится).
+        assert _canonical_room_bucket("Пентхаус") == "пентхаус"
+        assert _canonical_room_bucket(" Loft ") == "loft"
+
+
+class TestNormalizeSharesCanon:
+    def test_folds_duplicate_canon_keys(self) -> None:
+        # «studio»+«студия» схлопываются в один канон с суммой долей.
+        out = _normalize_shares({"studio": 0.25, "студия": 0.25, "1k": 0.5})
+        assert out is not None
+        assert set(out.keys()) == {"studio", "1k"}
+        assert out["studio"] == pytest.approx(0.5)
+        assert out["1k"] == pytest.approx(0.5)
+
+    def test_normalizes_to_sum_one(self) -> None:
+        out = _normalize_shares({"Студии 15-30": 30.0, "1-к 30-45": 10.0})
+        assert out is not None
+        assert sum(out.values()) == pytest.approx(1.0)
+        assert out["studio"] == pytest.approx(0.75)
+
+    def test_keys_sorted_deterministic(self) -> None:
+        out = _normalize_shares({"2k": 0.3, "studio": 0.3, "1k": 0.4})
+        assert out is not None
+        assert list(out.keys()) == sorted(out.keys())
+
+    def test_nonpositive_and_none_dropped(self) -> None:
+        out = _normalize_shares({"studio": 1.0, "1k": 0.0, "2k": -0.5})
+        assert out is not None
+        assert set(out.keys()) == {"studio"}
+
+    def test_empty_or_all_invalid_is_none(self) -> None:
+        assert _normalize_shares({}) is None
+        assert _normalize_shares({"studio": 0.0, "1k": -1.0}) is None
+
+
+class TestCandidateUnitMixFromRecommend:
+    def test_extracts_bucket_share_pairs(self) -> None:
+        # recommend_mix-форма: buckets=[{bucket, share_pct, …}] → {bucket: share}.
+        result = {
+            "buckets": [
+                {"bucket": "Студии 15-30", "share_pct": 30.0, "deal_count": 100},
+                {"bucket": "1-к 30-45", "share_pct": 45.0, "deal_count": 150},
+                {"bucket": "2-к 45-60", "share_pct": 25.0, "deal_count": 80},
+            ]
+        }
+        mix = _candidate_unit_mix_from_recommend(result)
+        assert mix == {"Студии 15-30": 30.0, "1-к 30-45": 45.0, "2-к 45-60": 25.0}
+
+    def test_skips_nonpositive_and_malformed(self) -> None:
+        result = {
+            "buckets": [
+                {"bucket": "Студии 15-30", "share_pct": 50.0},
+                {"bucket": "1-к 30-45", "share_pct": 0.0},  # 0 доля — пропуск
+                {"bucket": "2-к 45-60"},  # нет share_pct — пропуск
+                {"share_pct": 10.0},  # нет bucket — пропуск
+                "garbage",  # не dict — пропуск
+            ]
+        }
+        mix = _candidate_unit_mix_from_recommend(result)
+        assert mix == {"Студии 15-30": 50.0}
+
+    def test_empty_buckets_is_none(self) -> None:
+        assert _candidate_unit_mix_from_recommend({"buckets": []}) is None
+
+    def test_missing_buckets_key_is_none(self) -> None:
+        # district unknown / degraded recommend_mix без списка buckets → None.
+        assert _candidate_unit_mix_from_recommend({"scope": {"error": "x"}}) is None
+
+    def test_buckets_not_list_is_none(self) -> None:
+        assert _candidate_unit_mix_from_recommend({"buckets": "nope"}) is None
+
+    def test_extracted_mix_feeds_similarity_against_manual_own(self) -> None:
+        # end-to-end извлечения: extracted candidate ↔ manual own → ось похожести считается.
+        result = {
+            "buckets": [
+                {"bucket": "Студии 15-30", "share_pct": 50.0},
+                {"bucket": "1-к 30-45", "share_pct": 50.0},
+            ]
+        }
+        candidate = _candidate_unit_mix_from_recommend(result)
+        own = {"studio": 0.5, "1k": 0.5}
+        assert _unit_mix_similarity(candidate, own) == pytest.approx(1.0)
+
 
 class TestTimingOverlap:
     def test_same_month_full(self) -> None:
@@ -1608,3 +1739,363 @@ class TestCannibalizationDeterminism:
         can = _cannibalization_card(portfolio).indices[KEY_CANNIBALIZATION]
         names = [c["name"] for c in can.detail["top_contributors"]]
         assert names == ["Андрей", "Яков"]
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# §25.3 ось КВАРТИРОГРАФИИ — активируется из recommend_mix (4-я и последняя ось)
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _recommend_mix_result(shares_pct: dict[str, float]) -> dict[str, Any]:
+    """recommend_mix-подобный ответ: {bucket: share_pct} → форма с `buckets`."""
+    return {
+        "scope": {"district": "Академический"},
+        "buckets": [{"bucket": b, "share_pct": s} for b, s in shares_pct.items()],
+        "summary": {"warnings": []},
+    }
+
+
+def _unitmix_card(
+    portfolio: list[OwnProject],
+    *,
+    recommend_return: dict[str, Any] | None = None,
+    recommend_side_effect: Exception | None = None,
+    centroid: tuple[float, float] | None = _CENTROID,
+    cad_num: str | None = "66:41:0303161:123",
+) -> Any:
+    """compute_special_indices с замоканными own-portfolio + центроид + recommend_mix.
+
+    Дата отчёта фиксируется (_FixedDate), чтобы тайминговая ось была детерминирована и не
+    «шумела» при изоляции вклада квартирографии.
+    """
+    db = MagicMock()
+    rec_kwargs: dict[str, Any] = {}
+    if recommend_side_effect is not None:
+        rec_kwargs["side_effect"] = recommend_side_effect
+    else:
+        rec_kwargs["return_value"] = recommend_return
+    with (
+        _full_stack_patch(),
+        patch(f"{_MOD}.date", _FixedDate),
+        patch(f"{_MOD}.get_own_portfolio", return_value=portfolio),
+        patch(f"{_MOD}._query_parcel_centroid", return_value=centroid),
+        patch(f"{_MOD}.recommend_mix", MagicMock(**rec_kwargs)),
+        patch(
+            f"{_MOD}._query_artificial_demand",
+            return_value={"n_sold": 100, "n_mortgage": 40},
+        ),
+    ):
+        return compute_special_indices(
+            db, spec=_CAND_SPEC, district="Академический", cad_num=cad_num
+        )
+
+
+class TestCannibalizationUnitMixAxisFedFromRecommendMix:
+    """§25.3 ось квартирографии активируется из recommend_mix (4-я ось, follow-up)."""
+
+    def test_unit_mix_axis_contributes_when_recommend_resolves(self) -> None:
+        # recommend_mix отдаёт микс кандидата (RU-подписи), наш проект — manual латиница с
+        # ТЕМ ЖЕ распределением по комнатности → ось квартирографии = 1.0 (через канон).
+        own = _own(
+            "Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+            unit_mix={"studio": 0.3, "1k": 0.4, "2k": 0.3},
+        )
+        card = _unitmix_card(
+            [own],
+            recommend_return=_recommend_mix_result(
+                {"Студии 15-30": 30.0, "1-к 30-45": 40.0, "2-к 45-60": 30.0}
+            ),
+        )
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.detail["axes_available"]["unit_mix"] == 1  # ось активна (НЕ 0)
+        top = can.detail["top_contributors"][0]
+        assert top["axes"]["unit_mix"] == pytest.approx(1.0)  # идентичный микс по канону
+
+    def test_similar_mix_scores_higher_than_dissimilar(self) -> None:
+        # тот же recommend-микс кандидата; наш проект с ПОХОЖИМ vs НЕпохожим миксом.
+        # Все прочие оси (класс/цена/гео/тайминг) идентичны → разница только в квартирографии.
+        rec = _recommend_mix_result({"Студии 15-30": 50.0, "1-к 30-45": 50.0})
+        similar = _unitmix_card(
+            [_own("Похожий", lon=_CENTROID[0], lat=_CENTROID[1],
+                  unit_mix={"studio": 0.5, "1k": 0.5})],
+            recommend_return=rec,
+        )
+        dissimilar = _unitmix_card(
+            [_own("Непохожий", lon=_CENTROID[0], lat=_CENTROID[1],
+                  unit_mix={"2k": 0.5, "3k": 0.5})],
+            recommend_return=rec,
+        )
+        sim_axis = similar.indices[KEY_CANNIBALIZATION].detail[
+            "top_contributors"
+        ][0]["axes"]["unit_mix"]
+        dis_axis = dissimilar.indices[KEY_CANNIBALIZATION].detail[
+            "top_contributors"
+        ][0]["axes"]["unit_mix"]
+        assert sim_axis > dis_axis
+        # и итоговое значение каннибализации выше при похожем миксе.
+        assert (
+            similar.indices[KEY_CANNIBALIZATION].value
+            > dissimilar.indices[KEY_CANNIBALIZATION].value
+        )
+
+    def test_axis_excluded_when_recommend_returns_none(self) -> None:
+        # recommend_mix вернул None → микс None → ось квартирографии исключена (None-not-0),
+        # каннибализация всё равно считается по классу/цене/тайм/гео.
+        own = _own(
+            "Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+            unit_mix={"studio": 0.5, "1k": 0.5},
+        )
+        card = _unitmix_card([own], recommend_return=None)
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.method == "own_portfolio_overlap"
+        assert can.value is not None
+        assert can.detail["axes_available"]["unit_mix"] == 0  # НЕ сфабрикована
+        assert can.detail["top_contributors"][0]["axes"]["unit_mix"] is None
+
+    def test_axis_excluded_when_recommend_empty_buckets(self) -> None:
+        # recommend_mix отдал пустые buckets (тонкие данные) → микс None → ось исключена.
+        own = _own(
+            "Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+            unit_mix={"studio": 0.5, "1k": 0.5},
+        )
+        card = _unitmix_card([own], recommend_return={"buckets": []})
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.value is not None
+        assert can.detail["axes_available"]["unit_mix"] == 0
+
+    def test_axis_excluded_when_recommend_raises_no_crash(self) -> None:
+        # recommend_mix БРОСИЛ → graceful None → ось исключена, карточка цела (НЕ crash).
+        own = _own(
+            "Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+            unit_mix={"studio": 0.5, "1k": 0.5},
+        )
+        card = _unitmix_card([own], recommend_side_effect=RuntimeError("heavy query boom"))
+        assert len(card.indices) == 6  # карточка возвращена целиком
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.method == "own_portfolio_overlap"
+        assert can.value is not None
+        assert can.detail["axes_available"]["unit_mix"] == 0
+
+    def test_axis_excluded_when_own_has_no_unit_mix(self) -> None:
+        # ЧЕСТНОСТЬ (own_portfolio PR1): current-проект несёт unit_mix=None → даже при
+        # валидном recommend-миксе кандидата ось не считается (нечего сравнивать).
+        own = _own(
+            "Текущий", source="current", lon=_CENTROID[0], lat=_CENTROID[1],
+            unit_mix=None,
+        )
+        card = _unitmix_card(
+            [own],
+            recommend_return=_recommend_mix_result({"Студии 15-30": 100.0}),
+        )
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.detail["axes_available"]["unit_mix"] == 0
+        assert can.detail["top_contributors"][0]["axes"]["unit_mix"] is None
+
+    def test_all_four_axes_active_together(self) -> None:
+        # class + price + timing + unit_mix ВСЕ доступны на одном проекте → n_axes == 4.
+        own = _own(
+            "Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+            unit_mix={"studio": 0.5, "1k": 0.5},
+            release_month=_DERIVED_CANDIDATE_MONTH,  # совпадает с выведенным тайм. кандидата
+        )
+        card = _unitmix_card(
+            [own],
+            recommend_return=_recommend_mix_result({"Студии 15-30": 50.0, "1-к 30-45": 50.0}),
+        )
+        can = card.indices[KEY_CANNIBALIZATION]
+        summary = can.detail["axes_available"]
+        assert summary == {"class": 1, "price": 1, "unit_mix": 1, "timing": 1}
+        assert can.detail["top_contributors"][0]["n_axes"] == 4
+
+    def test_unit_mix_axis_deterministic_identical_as_dict(self) -> None:
+        # Детерминизм (§16): одинаковые входы → идентичный as_dict (с активной осью).
+        portfolio = [
+            _own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+                 unit_mix={"studio": 0.3, "1k": 0.4, "2k": 0.3}),
+            _own("Наш-Б", obj_class="комфорт+", lon=60.65, lat=56.85,
+                 unit_mix={"1k": 0.6, "2k": 0.4}),
+        ]
+        rec = _recommend_mix_result(
+            {"Студии 15-30": 30.0, "1-к 30-45": 40.0, "2-к 45-60": 30.0}
+        )
+        first = _unitmix_card(list(portfolio), recommend_return=rec).indices[
+            KEY_CANNIBALIZATION
+        ].as_dict()
+        second = _unitmix_card(list(portfolio), recommend_return=rec).indices[
+            KEY_CANNIBALIZATION
+        ].as_dict()
+        assert first == second
+        # подтверждаем, что ось реально участвовала (не пустой детерминизм).
+        assert first["detail"]["axes_available"]["unit_mix"] == 2
+
+    def test_recommend_mix_called_with_expected_args(self) -> None:
+        # cost/корректность: recommend_mix зовётся с district+target_class+cad_num кандидата.
+        own = _own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+                   unit_mix={"studio": 1.0})
+        rec_mock = MagicMock(return_value=_recommend_mix_result({"Студии 15-30": 100.0}))
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(f"{_MOD}.date", _FixedDate),
+            patch(f"{_MOD}.get_own_portfolio", return_value=[own]),
+            patch(f"{_MOD}._query_parcel_centroid", return_value=_CENTROID),
+            patch(f"{_MOD}.recommend_mix", rec_mock),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            compute_special_indices(
+                db, spec=_CAND_SPEC, district="Академический", cad_num="66:41:0303161:123"
+            )
+        assert rec_mock.call_count == 1  # ровно один доп. вызов на отчёт (не O(n))
+        _, kwargs = rec_mock.call_args
+        assert kwargs["district_name"] == "Академический"
+        assert kwargs["target_class"] == "комфорт"  # из _CAND_SPEC.obj_class
+        assert kwargs["cad_num"] == "66:41:0303161:123"
+
+    def test_no_district_skips_recommend_mix(self) -> None:
+        # нет района → recommend_mix НЕ зовётся (район обязателен) → ось исключена, без вызова.
+        own = _own("Наш-А", lon=_CENTROID[0], lat=_CENTROID[1],
+                   unit_mix={"studio": 1.0})
+        rec_mock = MagicMock(return_value=_recommend_mix_result({"Студии 15-30": 100.0}))
+        db = MagicMock()
+        with (
+            _full_stack_patch(),
+            patch(f"{_MOD}.date", _FixedDate),
+            patch(f"{_MOD}.get_own_portfolio", return_value=[own]),
+            patch(f"{_MOD}._query_parcel_centroid", return_value=_CENTROID),
+            patch(f"{_MOD}.recommend_mix", rec_mock),
+            patch(
+                f"{_MOD}._query_artificial_demand",
+                return_value={"n_sold": 100, "n_mortgage": 40},
+            ),
+        ):
+            card = compute_special_indices(
+                db, spec=_CAND_SPEC, district=None, cad_num="66:41:0303161:123"
+            )
+        assert rec_mock.call_count == 0  # район обязателен → нет вызова
+        assert card.indices[KEY_CANNIBALIZATION].detail["axes_available"]["unit_mix"] == 0
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# §25.3 hot-path gate (#1129 regression): тяжёлый recommend_mix зовётся ТОЛЬКО когда
+# в портфеле есть проект с unit_mix; get_own_portfolio фетчится РОВНО один раз за отчёт.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _gated_card(
+    portfolio: list[OwnProject],
+) -> tuple[Any, MagicMock, MagicMock]:
+    """Прогнать compute_special_indices, вернув (card, recommend_mix_mock, portfolio_mock).
+
+    Оба зависимых сервиса — явные MagicMock, чтобы тест проверял ИХ call_count: гейт
+    дёргает recommend_mix только при наличии unit_mix в портфеле, а get_own_portfolio
+    фетчится один раз (без двойного запроса). Дата отчёта зафиксирована (_FixedDate).
+    """
+    db = MagicMock()
+    rec_mock = MagicMock(return_value=_recommend_mix_result({"Студии 15-30": 100.0}))
+    portfolio_mock = MagicMock(return_value=portfolio)
+    with (
+        _full_stack_patch(),
+        patch(f"{_MOD}.date", _FixedDate),
+        patch(f"{_MOD}.get_own_portfolio", portfolio_mock),
+        patch(f"{_MOD}._query_parcel_centroid", return_value=_CENTROID),
+        patch(f"{_MOD}.recommend_mix", rec_mock),
+        patch(
+            f"{_MOD}._query_artificial_demand",
+            return_value={"n_sold": 100, "n_mortgage": 40},
+        ),
+    ):
+        card = compute_special_indices(
+            db, spec=_CAND_SPEC, district="Академический", cad_num="66:41:0303161:123"
+        )
+    return card, rec_mock, portfolio_mock
+
+
+class TestCannibalizationUnitMixGate:
+    """§25.3 hot-path gate — recommend_mix только при own-портфеле с миксом (#1129)."""
+
+    def test_no_project_with_unit_mix_skips_recommend_mix(self) -> None:
+        # Портфель только из current/domrf (unit_mix=None) → ось квартирографии всё равно
+        # исключилась бы → тяжёлый recommend_mix НЕ должен вызываться (call_count == 0).
+        portfolio = [
+            _own("Текущий-1", source="current", lon=_CENTROID[0], lat=_CENTROID[1],
+                 unit_mix=None, release_month=_DERIVED_CANDIDATE_MONTH),
+            _own("Текущий-2", source="current", lon=_CENTROID[0], lat=_CENTROID[1],
+                 unit_mix=None, release_month=_DERIVED_CANDIDATE_MONTH),
+        ]
+        card, rec_mock, _ = _gated_card(portfolio)
+        # Гейт сработал: тяжёлый запрос НЕ выполнен.
+        assert rec_mock.call_count == 0
+        # Ось квартирографии исключена (None-not-0), но каннибализация ВСЁ РАВНО считается
+        # по class+price+timing+geo (не падает, не деградирует целиком).
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.method == "own_portfolio_overlap"
+        assert can.value is not None
+        assert can.detail["axes_available"]["unit_mix"] == 0
+        assert can.detail["top_contributors"][0]["axes"]["unit_mix"] is None
+        # class+price+timing активны — каннибализация считается из остальных осей.
+        assert can.detail["axes_available"]["class"] == 2
+        assert can.detail["axes_available"]["price"] == 2
+        assert can.detail["axes_available"]["timing"] == 2
+
+    def test_project_with_unit_mix_calls_recommend_mix_once(self) -> None:
+        # Есть future-проект с unit_mix → ось может внести вклад → recommend_mix зовётся
+        # РОВНО один раз (не O(n), не дважды), и get_own_portfolio фетчится РОВНО один раз.
+        portfolio = [
+            _own("Будущий-А", source="future", lon=_CENTROID[0], lat=_CENTROID[1],
+                 unit_mix={"studio": 1.0}),
+        ]
+        card, rec_mock, portfolio_mock = _gated_card(portfolio)
+        # Тяжёлый запрос — ровно один доп. вызов на отчёт.
+        assert rec_mock.call_count == 1
+        # get_own_portfolio фетчится ОДИН раз (gate + каннибализация реюзают список — нет
+        # двойного запроса к БД).
+        assert portfolio_mock.call_count == 1
+        # Ось квартирографии активна и вносит вклад.
+        can = card.indices[KEY_CANNIBALIZATION]
+        assert can.detail["axes_available"]["unit_mix"] == 1
+        assert can.detail["top_contributors"][0]["axes"]["unit_mix"] == pytest.approx(1.0)
+
+    def test_get_own_portfolio_single_fetch_even_when_gate_skips(self) -> None:
+        # Даже когда гейт пропускает recommend_mix, get_own_portfolio всё равно фетчится
+        # РОВНО один раз (портфель нужен и для гейта, и для самой каннибализации).
+        portfolio = [
+            _own("Текущий-1", source="current", lon=_CENTROID[0], lat=_CENTROID[1],
+                 unit_mix=None),
+        ]
+        _, rec_mock, portfolio_mock = _gated_card(portfolio)
+        assert rec_mock.call_count == 0
+        assert portfolio_mock.call_count == 1  # один фетч, не ноль и не два
+
+    def test_empty_mix_dict_does_not_trigger_recommend_mix(self) -> None:
+        # Пустой dict unit_mix={} (грязь) — НЕ «есть микс» → recommend_mix не зовётся.
+        portfolio = [
+            _own("Будущий-пустой", source="future", lon=_CENTROID[0], lat=_CENTROID[1],
+                 unit_mix={}),
+        ]
+        _, rec_mock, _ = _gated_card(portfolio)
+        assert rec_mock.call_count == 0
+
+
+class TestPortfolioHasUnitMix:
+    """Pure-гейт _portfolio_has_unit_mix — есть ли проект с непустой квартирографией."""
+
+    def test_true_when_any_project_has_mix(self) -> None:
+        portfolio = [
+            _own("Без-микса", unit_mix=None),
+            _own("С-миксом", unit_mix={"studio": 0.5, "1k": 0.5}),
+        ]
+        assert _portfolio_has_unit_mix(portfolio) is True
+
+    def test_false_when_all_none(self) -> None:
+        portfolio = [_own("A", unit_mix=None), _own("B", unit_mix=None)]
+        assert _portfolio_has_unit_mix(portfolio) is False
+
+    def test_false_on_empty_portfolio(self) -> None:
+        assert _portfolio_has_unit_mix([]) is False
+
+    def test_empty_mix_dict_is_not_a_mix(self) -> None:
+        # пустой dict (грязь/нет долей) не считается миксом (falsy).
+        assert _portfolio_has_unit_mix([_own("A", unit_mix={})]) is False