backend/app/services/forecasting/__init__.py

@@ -9,7 +9,7 @@
   • macro_series  (#951b)          — monthly макро-ряд + классификатор режима ставки (X-ось §9.6).
   • sales_series  (#951c)          — monthly ряд продаж по сегменту (Y-ось §9.6).
   • rate_sensitivity (#951d)       — §9.6 чувствительность продаж к key_rate (CORE, ADVISORY).
-  • macro_coefficient (later PR)   — §9.5 макро-коэффициент.
+  • macro_coefficient (#951e)      — §9.5 макро-коэффициент (композитный множитель, ADVISORY).
 
 Источники данных:
   • макро — таблица macro_indicator через reader site_finder/macro.py (reuse).
@@ -18,6 +18,17 @@
 
 from __future__ import annotations
 
+from app.services.forecasting.macro_coefficient import (
+    MacroCoefficient,
+    assemble_coefficient,
+    compute_macro_coefficient,
+    f_issuance,
+    f_mortgage_rate,
+    f_overdue,
+    f_rate,
+    renormalize_contributions,
+    segment_steepness,
+)
 from app.services.forecasting.macro_series import (
     MonthlyMacro,
     classify_regime,
@@ -43,14 +54,21 @@ from app.services.forecasting.sales_series import (
 )
 
 __all__ = [
+    "MacroCoefficient",
     "MonthlyMacro",
     "RateSensitivity",
     "SalesSeries",
     "SegmentSpec",
+    "assemble_coefficient",
     "best_lag",
     "build_sales_series",
     "classify_regime",
+    "compute_macro_coefficient",
     "compute_rate_sensitivity",
+    "f_issuance",
+    "f_mortgage_rate",
+    "f_overdue",
+    "f_rate",
     "fill_month_grid",
     "get_monthly_macro",
     "is_confounded_window",
@@ -58,6 +76,8 @@ __all__ = [
     "macro_at_lag",
     "ols_slope_r2",
     "price_bucket_of",
+    "renormalize_contributions",
     "room_area_bucket_of",
+    "segment_steepness",
     "shrink",
 ]

backend/app/services/forecasting/macro_coefficient.py

@@ -0,0 +1,778 @@
+"""§9.5 макроэкономический коэффициент — детерминированный композитный множитель.
+
+#951 (Site Finder v2 / «GG-форсайт» ТЗ §9.5), EPIC 7 «Чувствительность к ключевой
+ставке», sub-PR 4 (#951e). §9.5 поправляет прогноз спроса на текущий МАКРО-РЕЖИМ:
+композитный множитель, центрированный в 1.0 (1.0 = «макрорежим нейтрален к спросу»;
+<1 = режим давит на спрос; >1 = режим поддерживает спрос), направленный по сегменту
+рынка. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (арифметика на именованных константах).
+
+ЭВРИСТИКА, НЕ ФИТ: веса и шаги-крутизны — ИМЕНОВАННЫЕ КОНСТАНТЫ (экспертная оценка),
+а НЕ обученная модель. Поэтому модуль уезжает в прод сейчас и уточняется позже на
+out-of-sample (как и §9.6 rate_sensitivity — тот тоже advisory до бэктеста PR6).
+
+ADVISORY-СТАТУС: коэффициент СОВЕТУЮЩИЙ и НЕ подключён ни к одному эндпоинту в этом
+PR. Цифры пригодны для explainability/прототипа, но НЕ для production-решений, пока
+веса не валидированы.
+
+Формула (центр 1.0; каждый sub-factor — направленный нудж в [-1, 1] ДО веса):
+
+    macro_coef = clamp(
+        1.0
+        + W_RATE       · f_rate(тренд key_rate)               # rate↑ → <0
+        + W_MORTG_RATE · f_mortgage_rate(Δ mortgage_rate_weighted)  # ↑ → <0
+        + W_ISSUANCE   · f_issuance(Δ count & Δ volume ИЖК)   # выдачи↓ → <0
+        + W_OVERDUE    · f_overdue(overdue / debt)            # доля↑ → <0
+        + W_GOVSUPPORT · None   # DEGRADED — нет поля в MonthlyMacro → не участвует
+        + W_INCOME     · None   # DEGRADED
+        + W_INFLATION  · None   # DEGRADED
+        + W_CONFIDENCE · None,  # DEGRADED
+        _MACRO_COEF_MIN, _MACRO_COEF_MAX)
+
+РЕНОРМАЛИЗАЦИЯ ВЕСОВ (критично): деградированные/None-входы выпадают И из числителя,
+И из суммы весов. Доступные веса делятся на свою сумму (renorm) → пропуск данных НЕ
+сжимает коэффициент к 1.0 искусственно (иначе 4 из 8 факторов = None всегда тянули
+бы coef к нейтрали и недооценивали реальный макрорежим). Все входы None → coef=1.0
+честно (нейтрально, нет сигнала), confidence='low'.
+
+СЕГМЕНТНЫЕ МОДИФИКАТОРЫ (§9.5 «особенно X»): rate-driven sub-factors (f_rate +
+f_mortgage_rate — оба про стоимость заёмных денег) масштабируются под профиль
+сегмента. Крупный / дорогой / высокочековый / инвестиционный формат реагирует
+КРУЧЕ-негативно на rate↑ (ипотечно-чувствителен); семейный / компактный / ликвидный
+comfort/comfort+ / быстрой нарезки — КРУЧЕ-позитивно на rate↓ (+ господдержка).
+Реализовано простым набором правил с именованными константами-крутизнами (НЕ фит).
+
+BACKED vs DEGRADED входы (сверено с полями MonthlyMacro PR2):
+  • BACKED   : f_rate (key_rate), f_mortgage_rate (mortgage_rate_weighted),
+               f_issuance (mortgage_issued_count + _volume), f_overdue
+               (mortgage_overdue / mortgage_debt).
+  • DEGRADED : gov-support / income / cpi / confidence — этих рядов в MonthlyMacro
+               НЕТ (income/cpi не backfilled, см. macro_series docstring; gov-support
+               и consumer-confidence вообще не собираются). Они зарезервированы в
+               формуле/весах для будущего, но СЕЙЧАС всегда contribute None →
+               выпадают из renorm. Помечаем их явно unavailable, не молчаливый 0.
+
+Graceful-on-thin-data (дух market_metrics.py / rate_sensitivity.py): пустой макро-ряд
+/ все входы None → coefficient=1.0 (нейтрально), confidence='low', НЕ crash.
+Детерминированно. numpy НЕ нужен (чистая арифметика).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from sqlalchemy.orm import Session
+
+from app.services.forecasting.macro_series import (
+    MonthlyMacro,
+    get_monthly_macro,
+    is_confounded_window,
+)
+
+logger = logging.getLogger(__name__)
+
+Confidence = Literal["high", "medium", "low"]
+
+# ── Named-константы: окно / клэмп ─────────────────────────────────────────────
+
+# Глубина макро-ряда (месяцев назад). 12 мес = «макрорежим за последний год»: §9.5
+# поправляет прогноз на ТЕКУЩИЙ режим, а не на 4-летний цикл (в отличие от §9.6
+# регрессии, которой нужен длинный ряд). Короче окно → свежее «фото» режима.
+_DEFAULT_MONTHS_BACK: int = 12
+
+# Клэмп-полоса множителя. Центр 1.0; асимметрична намеренно: вниз сильнее (0.6 =
+# режим может срезать прогноз спроса до −40%, как в острый rate-шок), вверх мягче
+# (1.3 = +30% максимум, господдержка/смягчение не «разгоняют» спрос так же резко,
+# как ужесточение его душит — асимметрия рынка ИЖК). Эвристика, уточняется бэктестом.
+_MACRO_COEF_MIN: float = 0.6
+_MACRO_COEF_MAX: float = 1.3
+_MACRO_COEF_NEUTRAL: float = 1.0
+
+# Окно (мес) для тренда/дельты внутри ряда: сравниваем последнее значение со
+# значением N месяцев назад. 6 ≈ полугодие — гасит месячный шум, но ловит разворот
+# режима (зеркалит дух лагов §9.6, где полугодовой лаг ловит ипотечный эффект).
+_TREND_WINDOW_MONTHS: int = 6
+
+# ── Named-константы: веса sub-factors (СУММА backed-весов = 0.45) ──────────────
+# Веса — экспертная оценка вклада каждого канала в макрорежим спроса (НЕ фит).
+# Заданы в ИСХОДНОМ (полном) наборе из 8 каналов; renorm делит на сумму ДОСТУПНЫХ.
+# Backed-каналы (rate/mortgage_rate/issuance/overdue) несут основную массу: ставка и
+# стоимость/доступность ипотеки — доминирующий драйвер первичного спроса в РФ.
+# Degraded-каналы (gov/income/cpi/confidence) имеют НЕнулевые веса в схеме (резерв
+# под будущие ряды), но СЕЙЧАС всегда None → в renorm не попадают.
+_W_RATE: float = 0.18  # ключевая ставка — главный канал (стоимость денег)
+_W_MORTG_RATE: float = 0.12  # ИЖК-ставка — прямой канал доступности ипотеки
+_W_ISSUANCE: float = 0.10  # объём/число выдач ИЖК — фактический спрос-прокси
+_W_OVERDUE: float = 0.05  # просрочка/долг — стресс заёмщиков (медленный канал)
+_W_GOVSUPPORT: float = 0.20  # DEGRADED (нет ряда) — крупный вес-резерв (льготы)
+_W_INCOME: float = 0.12  # DEGRADED — реальные доходы
+_W_INFLATION: float = 0.08  # DEGRADED — ИПЦ
+_W_CONFIDENCE: float = 0.15  # DEGRADED — потребит. уверенность
+
+# Имена sub-factors (ключи breakdown / available / weights). Держим как константы,
+# чтобы переименование не рассинхронило веса и breakdown молча.
+_F_RATE: str = "rate"
+_F_MORTG_RATE: str = "mortgage_rate"
+_F_ISSUANCE: str = "issuance"
+_F_OVERDUE: str = "overdue"
+_F_GOVSUPPORT: str = "gov_support"
+_F_INCOME: str = "income"
+_F_INFLATION: str = "inflation"
+_F_CONFIDENCE: str = "confidence"
+
+# Полная схема весов (исходный знаменатель ДО renorm). dict сохраняет порядок →
+# детерминированный обход breakdown.
+_WEIGHTS: dict[str, float] = {
+    _F_RATE: _W_RATE,
+    _F_MORTG_RATE: _W_MORTG_RATE,
+    _F_ISSUANCE: _W_ISSUANCE,
+    _F_OVERDUE: _W_OVERDUE,
+    _F_GOVSUPPORT: _W_GOVSUPPORT,
+    _F_INCOME: _W_INCOME,
+    _F_INFLATION: _W_INFLATION,
+    _F_CONFIDENCE: _W_CONFIDENCE,
+}
+
+# Каналы, для которых СЕЙЧАС нет ряда в MonthlyMacro → всегда contribute None.
+# Зарезервированы в _WEIGHTS под будущие данные; помечаем unavailable явно.
+_DEGRADED_FACTORS: frozenset[str] = frozenset(
+    {_F_GOVSUPPORT, _F_INCOME, _F_INFLATION, _F_CONFIDENCE}
+)
+
+# Rate-driven каналы — к ним применяется сегментный модификатор крутизны (оба про
+# стоимость заёмных денег: ставка ЦБ + ИЖК-ставка).
+_RATE_DRIVEN_FACTORS: frozenset[str] = frozenset({_F_RATE, _F_MORTG_RATE})
+
+# ── Named-константы: нормировка sub-factors (сырое Δ → нудж в [-1,1]) ──────────
+# Каждый sub-factor берёт сырую величину/дельту и сжимает в [-1, 1] делением на
+# «полную шкалу» (величина, при которой канал считаем максимально нагруженным) с
+# клэмпом. Шкалы — эмпирика РФ-рынка 2022-2024; эвристика, уточняется бэктестом.
+
+# Полная шкала тренда ключевой ставки (п.п. за окно). 8 п.п. ≈ «экстремальный»
+# разворот ДКП (напр. 7.5→20 в 2022 ушёл далеко за шкалу → нудж клэмпится в −1).
+_RATE_TREND_FULL_PP: float = 8.0
+
+# Полная шкала тренда ИЖК-ставки (п.п. за окно). 5 п.п. — резкое движение
+# средневзвешенной ИЖК-ставки (она инертнее ключевой, ходит меньшими шагами).
+_MORTG_RATE_TREND_FULL_PP: float = 5.0
+
+# Полная шкала относительного изменения выдач ИЖК за окно (доля). 0.5 = выдачи
+# упали/выросли вдвое (count или volume) → максимальный нудж. Берём СРЕДНЕЕ из
+# относительных Δ count и Δ volume (если оба есть; иначе доступный из двух).
+_ISSUANCE_REL_FULL: float = 0.5
+
+# Полная шкала доли просрочки (overdue/debt). 0.05 = 5% портфеля просрочено —
+# уже стрессовый уровень для ИЖК (исторически ИЖК-просрочка в РФ <1-2%). Это
+# УРОВЕНЬ (не дельта): высокая доля просрочки = плохой режим → негативный нудж.
+_OVERDUE_RATIO_FULL: float = 0.05
+# «Нейтральная» доля просрочки: ниже неё канал не давит (здоровый портфель). 1%.
+_OVERDUE_RATIO_NEUTRAL: float = 0.01
+
+# ── Named-константы: сегментные крутизны (множители rate-driven нуджей) ────────
+# Базовая крутизна 1.0 (нейтральный сегмент). Чувствительные форматы получают
+# >1.0 (круче реагируют на канал стоимости денег — в ОБЕ стороны: сильнее падают
+# при rate↑ и сильнее растут при rate↓). Эвристика «особенно X» §9.5.
+_STEEP_BASE: float = 1.0
+# Дорогой / бизнес-премиум / высокочековый: ипотека — меньшая доля сделки, но
+# инвест-мотив и крупный чек делают спрос капризнее к ставке. +40%.
+_STEEP_EXPENSIVE: float = 1.4
+# Крупный формат (3-к / 80+ м²): большой чек → почти всегда ипотека/длинный
+# горизонт → сильнее давит ставка. +30%.
+_STEEP_LARGE: float = 1.3
+# Инвестиционный мотив (покупка ради доходности, не для жизни): максимально
+# чувствителен к стоимости денег (альтернатива — депозит под высокую ставку). +50%.
+_STEEP_INVESTMENT: float = 1.5
+# Семейный / компактный / ликвидный comfort: массовый ипотечный спрос, который
+# СИЛЬНЕЕ оживает при смягчении + господдержке (адресат льготных программ). +25%.
+_STEEP_FAVORED: float = 1.25
+# Потолок суммарной крутизны (профиль может совпасть по нескольким признакам —
+# не даём множителям перемножиться в абсурд). Клэмпим итог в [мин, макс].
+_STEEP_MIN: float = 0.7
+_STEEP_MAX: float = 1.8
+
+# ── Named-константы: confidence ───────────────────────────────────────────────
+# Confidence падает, когда мало backed-входов доступно ИЛИ окно конфаундено.
+# 'high'   — доступны ВСЕ 4 backed-канала и окно не шок.
+# 'medium' — доступно ≥ _CONF_MEDIUM_MIN_BACKED backed-каналов (частичный сигнал).
+# 'low'    — почти нет сигнала (0-1 канал) ИЛИ деградация ниже medium-порога.
+_CONF_HIGH_MIN_BACKED: int = 4  # все backed-каналы
+_CONF_MEDIUM_MIN_BACKED: int = 2  # хотя бы половина backed-каналов
+
+
+@dataclass(frozen=True)
+class MacroCoefficient:
+    """§9.5 макроэкономический коэффициент — композитный множитель спроса.
+
+    Детерминированный результат. `coefficient` — клэмпнутый множитель в
+    [_MACRO_COEF_MIN, _MACRO_COEF_MAX], центр 1.0. `breakdown` — знаковый вклад
+    КАЖДОГО sub-factor В ИТОГ (после веса+renorm+сегмент-модификатора) ИЛИ None,
+    если канал недоступен (явно None, НИКОГДА 0-как-заглушка — пропуск ≠ нулевой
+    вклад). ADVISORY до валидации — не для production-решений.
+    """
+
+    coefficient: float
+    breakdown: dict[str, float | None]  # имя sub-factor → вклад в итог (None=недоступен)
+    available_inputs: list[str]  # backed-каналы, давшие сигнал
+    unavailable_inputs: list[str]  # деградированные / None-каналы
+    segment_profile: dict[str, Any]
+    confidence: Confidence
+    confounded: bool  # окно ряда пересекает шок-дату (PR2)
+    # renorm-множитель весов (доступная_сумма⁻¹·полная_сумма): для отладки/explain.
+    weight_renorm_factor: float | None = field(default=None)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "coefficient": _round_or_none(self.coefficient, 4),
+            "breakdown": {k: _round_or_none(v, 4) for k, v in self.breakdown.items()},
+            "available_inputs": list(self.available_inputs),
+            "unavailable_inputs": list(self.unavailable_inputs),
+            "segment_profile": dict(self.segment_profile),
+            "confidence": self.confidence,
+            "confounded": self.confounded,
+            "weight_renorm_factor": _round_or_none(self.weight_renorm_factor, 4),
+        }
+
+
+def _round_or_none(value: float | None, digits: int) -> float | None:
+    return round(value, digits) if value is not None else None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure-логика — без БД, полностью юнит-тестируемо (чистая арифметика).
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _clamp(value: float, lo: float, hi: float) -> float:
+    """Зажать value в [lo, hi]. PURE."""
+    return max(lo, min(hi, value))
+
+
+def f_rate(rate_trend_pp: float | None) -> float | None:
+    """Нудж от тренда ключевой ставки: rate↑ → негатив. None → недоступен.
+
+    rate_trend_pp = Δ(key_rate) за окно в п.п. (последнее − N мес назад). Растущая
+    ставка душит спрос → нудж < 0; падающая → нудж > 0. Сжимаем в [-1, 1] делением
+    на _RATE_TREND_FULL_PP с клэмпом (−знак: рост → отрицательный нудж). PURE.
+
+    Args:
+        rate_trend_pp: изменение ключевой ставки за окно (п.п.); None = нет данных.
+
+    Returns:
+        Нудж в [-1, 1] или None (вход None → канал недоступен).
+    """
+    if rate_trend_pp is None:
+        return None
+    return _clamp(-rate_trend_pp / _RATE_TREND_FULL_PP, -1.0, 1.0)
+
+
+def f_mortgage_rate(mortgage_rate_trend_pp: float | None) -> float | None:
+    """Нудж от тренда средневзвешенной ИЖК-ставки: ↑ → негатив. None → недоступен.
+
+    Прямой канал доступности ипотеки: рост ИЖК-ставки удорожает кредит → спрос ↓ →
+    нудж < 0. Сжимаем в [-1, 1] (_MORTG_RATE_TREND_FULL_PP). PURE.
+
+    Args:
+        mortgage_rate_trend_pp: Δ mortgage_rate_weighted за окно (п.п.); None = нет.
+
+    Returns:
+        Нудж в [-1, 1] или None.
+    """
+    if mortgage_rate_trend_pp is None:
+        return None
+    return _clamp(-mortgage_rate_trend_pp / _MORTG_RATE_TREND_FULL_PP, -1.0, 1.0)
+
+
+def f_issuance(count_rel_change: float | None, volume_rel_change: float | None) -> float | None:
+    """Нудж от динамики выдач ИЖК: выдачи↓ → негатив. Оба None → недоступен.
+
+    Относительное изменение за окно: rel = (последнее − база) / |база|. Падение
+    выдач (count и/или volume) = сжатие фактического спроса → нудж < 0; рост → > 0.
+    Берём СРЕДНЕЕ из доступных rel-Δ (count, volume); если доступен лишь один —
+    его. Сжимаем в [-1, 1] (_ISSUANCE_REL_FULL). PURE.
+
+    Args:
+        count_rel_change: относит. Δ числа выдач за окно; None = нет.
+        volume_rel_change: относит. Δ объёма выдач за окно; None = нет.
+
+    Returns:
+        Нудж в [-1, 1] или None (оба входа None → канал недоступен).
+    """
+    parts = [v for v in (count_rel_change, volume_rel_change) if v is not None]
+    if not parts:
+        return None
+    avg_rel = sum(parts) / len(parts)
+    return _clamp(avg_rel / _ISSUANCE_REL_FULL, -1.0, 1.0)
+
+
+def f_overdue(overdue: float | None, debt: float | None) -> float | None:
+    """Нудж от доли просрочки ИЖК (overdue/debt): доля↑ → негатив. None → недоступен.
+
+    УРОВЕНЬ (не дельта): высокая доля просрочки в портфеле = стресс заёмщиков =
+    плохой режим спроса → нудж < 0. Ниже _OVERDUE_RATIO_NEUTRAL (здоровый портфель)
+    канал не давит (нудж 0). От neutral до _OVERDUE_RATIO_FULL линейно уходит к −1.
+    Доля ВСЕГДА ≥ 0 → этот канал даёт ТОЛЬКО неположительный нудж (просрочка не
+    «помогает» спросу). PURE.
+
+    Args:
+        overdue: просроченная задолженность ИЖК (млн руб); None = нет.
+        debt: общая задолженность ИЖК (млн руб); None / ≤0 → недоступен (нет базы).
+
+    Returns:
+        Нудж в [-1, 0] или None (нет данных / нулевой портфель).
+    """
+    if overdue is None or debt is None or debt <= 0:
+        return None
+    ratio = overdue / debt
+    excess = ratio - _OVERDUE_RATIO_NEUTRAL
+    if excess <= 0:
+        return 0.0  # здоровый портфель — канал доступен, но нейтрален
+    span = _OVERDUE_RATIO_FULL - _OVERDUE_RATIO_NEUTRAL
+    return _clamp(-excess / span, -1.0, 0.0)
+
+
+def _trend_over_window(
+    values: list[float | None], *, window: int = _TREND_WINDOW_MONTHS
+) -> float | None:
+    """Абсолютный тренд ряда за окно: последнее_known − known_за_window_назад.
+
+    Берём последнюю непустую точку и точку примерно `window` индексов раньше неё
+    (первую непустую при поиске назад). Любая недоступна → None. Для key_rate /
+    mortgage_rate возвращает Δ в исходных единицах (п.п.). PURE.
+    """
+    last_idx = _last_known_idx(values)
+    if last_idx is None:
+        return None
+    base_idx = _known_idx_at_or_before(values, last_idx - window)
+    if base_idx is None or base_idx == last_idx:
+        return None
+    last_v = values[last_idx]
+    base_v = values[base_idx]
+    if last_v is None or base_v is None:  # отсечено выше, но страховка для mypy
+        return None
+    return float(last_v) - float(base_v)
+
+
+def _rel_change_over_window(
+    values: list[float | None], *, window: int = _TREND_WINDOW_MONTHS
+) -> float | None:
+    """Относительный тренд за окно: (последнее − база) / |база|. База ≤0 → None.
+
+    Для count/volume выдач (rel-Δ, безразмерный). Той же логикой поиска последней/
+    базовой точки, что _trend_over_window. PURE.
+    """
+    last_idx = _last_known_idx(values)
+    if last_idx is None:
+        return None
+    base_idx = _known_idx_at_or_before(values, last_idx - window)
+    if base_idx is None or base_idx == last_idx:
+        return None
+    last_v = values[last_idx]
+    base_v = values[base_idx]
+    if last_v is None or base_v is None or base_v == 0:
+        return None
+    return (float(last_v) - float(base_v)) / abs(float(base_v))
+
+
+def _last_known_idx(values: list[float | None]) -> int | None:
+    """Индекс последней непустой точки ряда (или None, если все None). PURE."""
+    for i in range(len(values) - 1, -1, -1):
+        if values[i] is not None:
+            return i
+    return None
+
+
+def _known_idx_at_or_before(values: list[float | None], idx: int) -> int | None:
+    """Индекс непустой точки на позиции idx или ближайшей раньше неё. PURE.
+
+    Идём от min(idx, last) назад до первой непустой. idx<0 → None. Так база тренда
+    устойчива к дыркам ровно на целевой позиции (берём ближайшую известную раньше).
+    """
+    if idx < 0:
+        return None
+    start = min(idx, len(values) - 1)
+    for i in range(start, -1, -1):
+        if values[i] is not None:
+            return i
+    return None
+
+
+def _latest_ratio(
+    numerators: list[float | None], denominators: list[float | None]
+) -> tuple[float | None, float | None]:
+    """Последняя пара (numerator, denominator) с обоими непустыми значениями.
+
+    Для f_overdue: берём свежайший месяц, где есть И overdue, И debt (доля просрочки
+    — про текущий УРОВЕНЬ, не тренд). Идём с конца. Нет такой пары → (None, None).
+    PURE.
+    """
+    n = min(len(numerators), len(denominators))
+    for i in range(n - 1, -1, -1):
+        num = numerators[i]
+        den = denominators[i]
+        if num is not None and den is not None:
+            return num, den
+    return None, None
+
+
+def segment_steepness(segment_profile: dict[str, Any]) -> float:
+    """Сегментная крутизна для rate-driven нуджей (§9.5 «особенно X»). PURE.
+
+    Берёт профиль сегмента и возвращает множитель крутизны в [_STEEP_MIN,
+    _STEEP_MAX], центр _STEEP_BASE=1.0. Чувствительные форматы (дорогой / крупный /
+    высокочековый / инвестиционный) → >1.0 (круче реагируют на канал стоимости
+    денег в ОБЕ стороны); favored массовый (семейный / компактный / ликвидный
+    comfort) → тоже >1.0, но через «оживание при смягчении».
+
+    Признаки профиля (все опциональны; неизвестное → не влияет):
+      • obj_class    : 'бизнес'/'премиум'/'элит…' → expensive; 'комфорт'/'comfort+'
+                       → favored; 'эконом' → favored (массовый ипотечный).
+      • price_tier   : 'high'/'premium'/'бизнес'/'премиум' → expensive;
+                       'low'/'эконом'/'comfort'… → favored.
+      • room_bucket  : крупная нарезка (3-к / 80+ / '4'/'5+') → large; компактная
+                       (студия / 1-к) → favored.
+      • is_investment: True → investment (максимальная крутизна).
+      • is_family / is_compact / is_liquid : True → favored.
+
+    Реализация: стартуем с _STEEP_BASE, домножаем на каждую сработавшую крутизну
+    (нормированную к base), клэмпим итог. Простой набор правил, НЕ фит.
+
+    Args:
+        segment_profile: словарь признаков сегмента (любой subset).
+
+    Returns:
+        Множитель крутизны в [_STEEP_MIN, _STEEP_MAX].
+    """
+    steep = _STEEP_BASE
+    for factor in _segment_steepness_factors(segment_profile):
+        # Домножаем на относительную крутизну (factor/base), чтобы база 1.0 была
+        # нейтральна, а несколько признаков усиливали друг друга (с потолком).
+        steep *= factor / _STEEP_BASE
+    return _clamp(steep, _STEEP_MIN, _STEEP_MAX)
+
+
+def _segment_steepness_factors(profile: dict[str, Any]) -> list[float]:
+    """Список сработавших крутизн из профиля (порядок детерминирован). PURE."""
+    factors: list[float] = []
+
+    obj_class = _norm_str(profile.get("obj_class"))
+    price_tier = _norm_str(profile.get("price_tier"))
+    room_bucket = _norm_str(profile.get("room_bucket"))
+
+    if _is_expensive_class(obj_class) or _is_expensive_tier(price_tier):
+        factors.append(_STEEP_EXPENSIVE)
+    if _is_large_room(room_bucket):
+        factors.append(_STEEP_LARGE)
+    if _truthy(profile.get("is_investment")):
+        factors.append(_STEEP_INVESTMENT)
+    if (
+        _is_favored_class(obj_class)
+        or _is_favored_tier(price_tier)
+        or _is_compact_room(room_bucket)
+        or _truthy(profile.get("is_family"))
+        or _truthy(profile.get("is_compact"))
+        or _truthy(profile.get("is_liquid"))
+    ):
+        factors.append(_STEEP_FAVORED)
+    return factors
+
+
+def _norm_str(value: Any) -> str | None:
+    """Привести значение к lowercase-строке (для регистро-независимого матча). PURE."""
+    if value is None:
+        return None
+    return str(value).strip().lower()
+
+
+def _truthy(value: Any) -> bool:
+    """True только для явного булева True (None/0/'' → False). PURE."""
+    return value is True
+
+
+# Подстроки классов/тиров (lowercase) — РФ-нейминг + латиница. Подстрочный матч,
+# чтобы 'премиум-плюс'/'business' и т.п. попадали без точного перечисления.
+_EXPENSIVE_CLASS_MARKERS: tuple[str, ...] = (
+    "бизнес",
+    "business",
+    "премиум",
+    "premium",
+    "элит",
+    "elit",
+)
+_FAVORED_CLASS_MARKERS: tuple[str, ...] = (
+    "комфорт",
+    "comfort",
+    "эконом",
+    "econom",
+    "стандарт",
+    "standard",
+)
+_EXPENSIVE_TIER_MARKERS: tuple[str, ...] = ("high", "premium", "бизнес", "премиум", "верхн")
+_FAVORED_TIER_MARKERS: tuple[str, ...] = (
+    "low",
+    "mid",
+    "эконом",
+    "комфорт",
+    "comfort",
+    "нижн",
+    "средн",
+)
+# Крупная нарезка: 3-к и больше / явные крупные метки room_area_bucket / Source A.
+_LARGE_ROOM_MARKERS: tuple[str, ...] = ("3-к", "80+", "4-к", "5+", "5-к")
+_COMPACT_ROOM_MARKERS: tuple[str, ...] = ("студи", "1-к", "studio")
+
+
+def _has_marker(value: str | None, markers: tuple[str, ...]) -> bool:
+    """True, если value содержит любую из подстрок-маркеров. PURE."""
+    if value is None:
+        return False
+    return any(m in value for m in markers)
+
+
+def _is_expensive_class(v: str | None) -> bool:
+    return _has_marker(v, _EXPENSIVE_CLASS_MARKERS)
+
+
+def _is_favored_class(v: str | None) -> bool:
+    return _has_marker(v, _FAVORED_CLASS_MARKERS)
+
+
+def _is_expensive_tier(v: str | None) -> bool:
+    return _has_marker(v, _EXPENSIVE_TIER_MARKERS)
+
+
+def _is_favored_tier(v: str | None) -> bool:
+    return _has_marker(v, _FAVORED_TIER_MARKERS)
+
+
+def _is_large_room(v: str | None) -> bool:
+    if v is None:
+        return False
+    if v in ("3", "4", "5", "4+", "5+"):  # Source A сырые room-ключи
+        return True
+    return _has_marker(v, _LARGE_ROOM_MARKERS)
+
+
+def _is_compact_room(v: str | None) -> bool:
+    if v is None:
+        return False
+    if v in ("0", "1", "студия"):  # Source A сырые room-ключи (0=студия)
+        return True
+    return _has_marker(v, _COMPACT_ROOM_MARKERS)
+
+
+def renormalize_contributions(
+    nudges: dict[str, float | None], weights: dict[str, float]
+) -> tuple[dict[str, float | None], float | None]:
+    """Свести нуджи sub-factors во взвешенные вклады с РЕНОРМАЛИЗАЦИЕЙ весов.
+
+    Доступные каналы (nudge != None) делят свой вес на СУММУ доступных весов →
+    итоговые веса доступных суммируются в полную исходную сумму весов (не в ≤1
+    из-за выпавших). Так пропуск данных НЕ тянет coef к 1.0 искусственно: каждый
+    оставшийся канал «забирает» долю выпавших пропорционально своему весу.
+
+    Renorm-множитель = (Σ всех весов) / (Σ доступных весов). Недоступные каналы →
+    вклад None (явно, не 0). PURE.
+
+    Args:
+        nudges: имя sub-factor → нудж в [-1,1] или None (недоступен).
+        weights: имя sub-factor → исходный вес (полная схема).
+
+    Returns:
+        (contributions, renorm_factor):
+          • contributions: имя → взвешенный вклад (renorm·вес·нудж) или None.
+          • renorm_factor: множитель ренормализации или None (нет доступных).
+    """
+    total_weight = sum(weights.values())
+    available_weight = sum(
+        weights.get(name, 0.0) for name, nudge in nudges.items() if nudge is not None
+    )
+    if available_weight <= 0:
+        # Ни одного доступного канала → все вклады None, renorm неопределён.
+        return {name: None for name in nudges}, None
+
+    renorm = total_weight / available_weight
+    contributions: dict[str, float | None] = {}
+    for name, nudge in nudges.items():
+        if nudge is None:
+            contributions[name] = None
+        else:
+            contributions[name] = renorm * weights.get(name, 0.0) * nudge
+    return contributions, renorm
+
+
+def assemble_coefficient(contributions: dict[str, float | None]) -> float:
+    """Собрать клэмпнутый множитель из взвешенных вкладов: 1.0 + Σ вкладов. PURE.
+
+    None-вклады (недоступные каналы) пропускаются. Итог клэмпится в
+    [_MACRO_COEF_MIN, _MACRO_COEF_MAX]. Пустые/все-None вклады → 1.0 (нейтрально).
+
+    Args:
+        contributions: имя sub-factor → взвешенный вклад или None.
+
+    Returns:
+        Множитель в [_MACRO_COEF_MIN, _MACRO_COEF_MAX].
+    """
+    total = _MACRO_COEF_NEUTRAL + sum(c for c in contributions.values() if c is not None)
+    return _clamp(total, _MACRO_COEF_MIN, _MACRO_COEF_MAX)
+
+
+def _confidence(*, n_backed_available: int, confounded: bool) -> Confidence:
+    """Confidence §9.5 по числу доступных backed-каналов + шок-окно. PURE.
+
+    'high'   — доступны ВСЕ backed-каналы и окно не конфаундено.
+    'medium' — доступно ≥ _CONF_MEDIUM_MIN_BACKED backed-каналов (частичный сигнал)
+               ЛИБО все backed есть, но окно конфаундено (шок снижает доверие).
+    'low'    — почти нет сигнала (< medium-порога доступных каналов).
+    """
+    if n_backed_available < _CONF_MEDIUM_MIN_BACKED:
+        return "low"
+    if n_backed_available >= _CONF_HIGH_MIN_BACKED and not confounded:
+        return "high"
+    return "medium"
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# DB-оркестратор — тонкий, graceful. Pure-логика выше тестируется без него.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def compute_macro_coefficient(
+    db: Session,
+    *,
+    segment_profile: dict[str, Any] | None = None,
+    months_back: int = _DEFAULT_MONTHS_BACK,
+) -> MacroCoefficient:
+    """Вычислить §9.5 макроэкономический коэффициент для сегмента.
+
+    ADVISORY (веса — эвристика, не фит): НЕ подключать в production-эндпоинт, пока
+    не валидировано. Множитель центрирован в 1.0; <1 = режим давит, >1 = поддержит.
+
+    Шаги:
+      1. Тянем monthly макро-ряд (PR2). Пустой → coef=1.0, confidence='low'.
+      2. Считаем 4 BACKED sub-factors (rate / mortgage_rate / issuance / overdue);
+         degraded каналы (gov/income/cpi/confidence) → None (нет ряда в MonthlyMacro).
+      3. Применяем сегментную крутизну к rate-driven нуджам (rate + mortgage_rate).
+      4. Ренормализуем веса по ДОСТУПНЫМ каналам (выпавшие не тянут coef к 1.0).
+      5. Собираем clamp(1.0 + Σ вкладов, MIN, MAX); confidence + confounded-флаг.
+
+    Graceful: пусто/все None/ошибка ряда → coef=1.0 (нейтрально), confidence='low',
+    НЕ crash. Детерминированно.
+
+    Args:
+        db: SQLAlchemy sync Session.
+        segment_profile: признаки сегмента (obj_class / price_tier / room_bucket /
+            is_investment / is_family / is_compact / is_liquid). None → нейтральный.
+        months_back: глубина макро-ряда (по умолчанию _DEFAULT_MONTHS_BACK).
+
+    Returns:
+        MacroCoefficient (всегда; coef=1.0 при нехватке данных).
+    """
+    profile = dict(segment_profile or {})
+
+    macro = get_monthly_macro(db, months_back=months_back)
+    confounded = _series_confounded(macro)
+
+    # ── 2. Сырые нуджи backed-каналов (degraded → None) ────────────────────────
+    base_nudges = _compute_backed_nudges(macro)
+
+    # ── 3. Сегментная крутизна на rate-driven нуджи ────────────────────────────
+    steepness = segment_steepness(profile)
+    nudges = _apply_segment_steepness(base_nudges, steepness)
+
+    # Degraded-каналы добавляем как явные None (для breakdown / unavailable-списка).
+    for name in _DEGRADED_FACTORS:
+        nudges.setdefault(name, None)
+
+    # ── 4. Ренормализация весов по доступным каналам ───────────────────────────
+    contributions, renorm = renormalize_contributions(nudges, _WEIGHTS)
+
+    # ── 5. Сборка + клэмп + confidence ─────────────────────────────────────────
+    coefficient = assemble_coefficient(contributions)
+
+    available = [name for name, nudge in nudges.items() if nudge is not None]
+    unavailable = [name for name, nudge in nudges.items() if nudge is None]
+    n_backed_available = sum(1 for name in available if name not in _DEGRADED_FACTORS)
+    confidence = _confidence(n_backed_available=n_backed_available, confounded=confounded)
+
+    logger.info(
+        "macro_coefficient: coef=%.4f backed_available=%d renorm=%s confounded=%s "
+        "steepness=%.3f confidence=%s segment=%s",
+        coefficient,
+        n_backed_available,
+        f"{renorm:.4f}" if renorm is not None else None,
+        confounded,
+        steepness,
+        confidence,
+        profile,
+    )
+
+    return MacroCoefficient(
+        coefficient=coefficient,
+        breakdown=contributions,
+        available_inputs=available,
+        unavailable_inputs=unavailable,
+        segment_profile=profile,
+        confidence=confidence,
+        confounded=confounded,
+        weight_renorm_factor=renorm,
+    )
+
+
+def _compute_backed_nudges(macro: list[MonthlyMacro]) -> dict[str, float | None]:
+    """Посчитать 4 backed sub-factors из макро-ряда (каждый None при нехватке). PURE-ish.
+
+    Не ходит в БД (получает уже вытянутый ряд) — тестируется напрямую.
+
+      • rate          : тренд key_rate за окно → f_rate.
+      • mortgage_rate : тренд mortgage_rate_weighted за окно → f_mortgage_rate.
+      • issuance      : относит. тренд count & volume выдач → f_issuance.
+      • overdue       : последняя доля overdue/debt → f_overdue.
+    """
+    key_rate = [m.key_rate for m in macro]
+    mortg_rate = [m.mortgage_rate_weighted for m in macro]
+    issued_count = [m.mortgage_issued_count for m in macro]
+    issued_volume = [m.mortgage_issued_volume for m in macro]
+    overdue_series = [m.mortgage_overdue for m in macro]
+    debt_series = [m.mortgage_debt for m in macro]
+
+    overdue_last, debt_last = _latest_ratio(overdue_series, debt_series)
+
+    return {
+        _F_RATE: f_rate(_trend_over_window(key_rate)),
+        _F_MORTG_RATE: f_mortgage_rate(_trend_over_window(mortg_rate)),
+        _F_ISSUANCE: f_issuance(
+            _rel_change_over_window(issued_count),
+            _rel_change_over_window(issued_volume),
+        ),
+        _F_OVERDUE: f_overdue(overdue_last, debt_last),
+    }
+
+
+def _apply_segment_steepness(
+    nudges: dict[str, float | None], steepness: float
+) -> dict[str, float | None]:
+    """Домножить rate-driven нуджи (rate + mortgage_rate) на сегментную крутизну.
+
+    После домножения переклэмпиваем в [-1, 1] (крутизна может выгнать за границу).
+    Не-rate-driven и None-нуджи не трогаем. Возвращает НОВЫЙ dict (не мутирует вход).
+    PURE.
+    """
+    out: dict[str, float | None] = {}
+    for name, nudge in nudges.items():
+        if nudge is not None and name in _RATE_DRIVEN_FACTORS:
+            out[name] = _clamp(nudge * steepness, -1.0, 1.0)
+        else:
+            out[name] = nudge
+    return out
+
+
+def _series_confounded(macro: list[MonthlyMacro]) -> bool:
+    """True, если окно макро-ряда [min..max] пересекает шок-дату (reuse PR2). PURE.
+
+    Пустой ряд → False (нет окна — нечего конфаундить).
+    """
+    if not macro:
+        return False
+    months = [m.month for m in macro]
+    return is_confounded_window(min(months), max(months))

backend/tests/services/forecasting/test_macro_coefficient.py

@@ -0,0 +1,548 @@
+"""Unit-тесты §9.5 макроэкономического коэффициента (#951e, ADVISORY).
+
+Чистые тесты — БЕЗ живой БД (арифметика на синтетике + мок PR2 get_monthly_macro):
+  • pure sub-factors (f_rate / f_mortgage_rate / f_issuance / f_overdue) — знак +
+    границы [-1,1] + None-вход → None (недоступен).
+  • renormalize_contributions — деградированные входы выпадают из числителя И суммы
+    весов; coef НЕ тянется к 1.0 искусственно; все-None → renorm None.
+  • segment_steepness — large/expensive/investment → круче (>1.0); favored → >1.0;
+    нейтральный/неизвестный → 1.0; клэмп.
+  • assemble_coefficient — клэмп на MIN/MAX; None-вклады пропускаются.
+  • compute_macro_coefficient (мок PR2): rate↑+issuance↓ → coef<1; rate↓+gov-favored
+    → coef>1 для favored-сегмента; graceful пусто → 1.0/low.
+
+ADVISORY-статус (веса — эвристика) проверяется на уровне поведения (центр 1.0,
+направленность, renorm, graceful).
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import math
+import os
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services.forecasting.macro_coefficient import (
+    _DEGRADED_FACTORS,
+    _F_ISSUANCE,
+    _F_MORTG_RATE,
+    _F_OVERDUE,
+    _F_RATE,
+    _MACRO_COEF_MAX,
+    _MACRO_COEF_MIN,
+    _MACRO_COEF_NEUTRAL,
+    _STEEP_BASE,
+    _STEEP_MAX,
+    _STEEP_MIN,
+    _WEIGHTS,
+    MacroCoefficient,
+    assemble_coefficient,
+    compute_macro_coefficient,
+    f_issuance,
+    f_mortgage_rate,
+    f_overdue,
+    f_rate,
+    renormalize_contributions,
+    segment_steepness,
+)
+from app.services.forecasting.macro_series import MonthlyMacro
+
+_MACRO = "app.services.forecasting.macro_coefficient.get_monthly_macro"
+
+
+def _months(n: int, *, end: dt.date | None = None) -> list[dt.date]:
+    """n подряд идущих 1-х чисел месяцев, заканчивая end (по умолчанию 2023-12)."""
+    end = end or dt.date(2023, 12, 1)
+    out: list[dt.date] = []
+    y, m = end.year, end.month
+    for _ in range(n):
+        out.append(dt.date(y, m, 1))
+        m -= 1
+        if m == 0:
+            m = 12
+            y -= 1
+    return list(reversed(out))
+
+
+def _macro(
+    months: list[dt.date],
+    *,
+    key_rate: list[float | None] | None = None,
+    mortgage_rate: list[float | None] | None = None,
+    issued_count: list[float | None] | None = None,
+    issued_volume: list[float | None] | None = None,
+    debt: list[float | None] | None = None,
+    overdue: list[float | None] | None = None,
+) -> list[MonthlyMacro]:
+    """Список MonthlyMacro; невыставленные ряды → все None (degraded-вход)."""
+    n = len(months)
+    none_n: list[float | None] = [None] * n
+    kr = key_rate if key_rate is not None else none_n
+    mr = mortgage_rate if mortgage_rate is not None else none_n
+    ic = issued_count if issued_count is not None else none_n
+    iv = issued_volume if issued_volume is not None else none_n
+    db_ = debt if debt is not None else none_n
+    od = overdue if overdue is not None else none_n
+    out: list[MonthlyMacro] = []
+    for i, month in enumerate(months):
+        out.append(
+            MonthlyMacro(
+                month=month,
+                key_rate=kr[i],
+                mortgage_rate_weighted=mr[i],
+                mortgage_issued_count=ic[i],
+                mortgage_issued_volume=iv[i],
+                mortgage_debt=db_[i],
+                mortgage_overdue=od[i],
+            )
+        )
+    return out
+
+
+# ── pure: f_rate ──────────────────────────────────────────────────────────────
+
+
+class TestFRate:
+    def test_rate_up_is_negative(self) -> None:
+        assert f_rate(4.0) is not None
+        assert f_rate(4.0) < 0  # ставка ↑ → давит спрос
+
+    def test_rate_down_is_positive(self) -> None:
+        v = f_rate(-4.0)
+        assert v is not None and v > 0  # ставка ↓ → поддержит спрос
+
+    def test_zero_trend_is_zero(self) -> None:
+        assert f_rate(0.0) == 0.0
+
+    def test_bounded_below_minus_one(self) -> None:
+        # Экстремальный рост (12 п.п. > full scale 8) → клэмп в −1.
+        assert f_rate(100.0) == -1.0
+
+    def test_bounded_above_plus_one(self) -> None:
+        assert f_rate(-100.0) == 1.0
+
+    def test_none_is_unavailable(self) -> None:
+        assert f_rate(None) is None
+
+
+# ── pure: f_mortgage_rate ─────────────────────────────────────────────────────
+
+
+class TestFMortgageRate:
+    def test_up_negative_down_positive(self) -> None:
+        up = f_mortgage_rate(3.0)
+        down = f_mortgage_rate(-3.0)
+        assert up is not None and up < 0
+        assert down is not None and down > 0
+
+    def test_bounds(self) -> None:
+        assert f_mortgage_rate(50.0) == -1.0
+        assert f_mortgage_rate(-50.0) == 1.0
+
+    def test_none_unavailable(self) -> None:
+        assert f_mortgage_rate(None) is None
+
+
+# ── pure: f_issuance ──────────────────────────────────────────────────────────
+
+
+class TestFIssuance:
+    def test_drop_is_negative(self) -> None:
+        v = f_issuance(-0.3, -0.3)  # выдачи упали → негатив
+        assert v is not None and v < 0
+
+    def test_growth_is_positive(self) -> None:
+        v = f_issuance(0.3, 0.3)
+        assert v is not None and v > 0
+
+    def test_averages_two_inputs(self) -> None:
+        # count +0.5, volume −0.5 → среднее 0 → нудж 0.
+        assert f_issuance(0.5, -0.5) == 0.0
+
+    def test_single_available_input(self) -> None:
+        # Доступен только volume → берём его (count None не обнуляет канал).
+        v = f_issuance(None, -0.25)
+        assert v is not None and v < 0
+
+    def test_bounds(self) -> None:
+        assert f_issuance(-5.0, -5.0) == -1.0
+        assert f_issuance(5.0, 5.0) == 1.0
+
+    def test_both_none_unavailable(self) -> None:
+        assert f_issuance(None, None) is None
+
+
+# ── pure: f_overdue ───────────────────────────────────────────────────────────
+
+
+class TestFOverdue:
+    def test_high_ratio_is_negative(self) -> None:
+        # overdue/debt = 30/1000 = 3% > neutral 1% → негатив.
+        v = f_overdue(30.0, 1000.0)
+        assert v is not None and v < 0
+
+    def test_healthy_portfolio_is_neutral_zero(self) -> None:
+        # 0.5% < neutral 1% → канал доступен, но нудж 0 (не давит).
+        assert f_overdue(5.0, 1000.0) == 0.0
+
+    def test_only_non_positive(self) -> None:
+        # Просрочка не «помогает» спросу: нудж всегда ≤ 0.
+        v = f_overdue(80.0, 1000.0)  # 8% > full 5% → клэмп −1
+        assert v == -1.0
+
+    def test_none_or_zero_debt_unavailable(self) -> None:
+        assert f_overdue(None, 1000.0) is None
+        assert f_overdue(10.0, None) is None
+        assert f_overdue(10.0, 0.0) is None  # нулевой портфель → нет базы
+
+
+# ── pure: segment_steepness ───────────────────────────────────────────────────
+
+
+class TestSegmentSteepness:
+    def test_neutral_profile_is_base(self) -> None:
+        assert segment_steepness({}) == _STEEP_BASE
+
+    def test_unknown_fields_ignored(self) -> None:
+        assert segment_steepness({"foo": "bar", "obj_class": None}) == _STEEP_BASE
+
+    def test_expensive_class_steeper(self) -> None:
+        assert segment_steepness({"obj_class": "Бизнес"}) > _STEEP_BASE
+
+    def test_premium_tier_steeper(self) -> None:
+        assert segment_steepness({"price_tier": "premium"}) > _STEEP_BASE
+
+    def test_large_room_steeper(self) -> None:
+        assert segment_steepness({"room_bucket": "3-к 60-80"}) > _STEEP_BASE
+        assert segment_steepness({"room_bucket": "4"}) > _STEEP_BASE  # Source A ключ
+
+    def test_investment_steeper(self) -> None:
+        assert segment_steepness({"is_investment": True}) > _STEEP_BASE
+
+    def test_favored_family_compact_steeper(self) -> None:
+        assert segment_steepness({"obj_class": "комфорт"}) > _STEEP_BASE
+        assert segment_steepness({"is_family": True}) > _STEEP_BASE
+        assert segment_steepness({"room_bucket": "студия"}) > _STEEP_BASE
+
+    def test_compound_profile_clamped_to_max(self) -> None:
+        # Дорогой + крупный + инвестиционный — перемножение крутизн упёрлось бы за
+        # потолок; клэмп держит в _STEEP_MAX.
+        steep = segment_steepness(
+            {"obj_class": "премиум", "room_bucket": "80+ м²", "is_investment": True}
+        )
+        assert steep == _STEEP_MAX
+
+    def test_within_bounds(self) -> None:
+        for prof in ({}, {"obj_class": "бизнес"}, {"is_investment": True}):
+            s = segment_steepness(prof)
+            assert _STEEP_MIN <= s <= _STEEP_MAX
+
+
+# ── pure: renormalize_contributions (РЕНОРМАЛИЗАЦИЯ) ───────────────────────────
+
+
+class TestRenormalize:
+    def test_unavailable_dropped_to_none(self) -> None:
+        nudges: dict[str, float | None] = {"a": -0.5, "b": None, "c": 0.2}
+        weights = {"a": 0.2, "b": 0.3, "c": 0.1}
+        contribs, renorm = renormalize_contributions(nudges, weights)
+        assert contribs["b"] is None  # недоступный → None, НЕ 0
+        assert contribs["a"] is not None and contribs["c"] is not None
+        assert renorm is not None
+
+    def test_renorm_factor_scales_by_dropped_weight(self) -> None:
+        # Доступны a(0.2)+c(0.1)=0.3 из total 0.6 → renorm = 0.6/0.3 = 2.0.
+        nudges: dict[str, float | None] = {"a": -0.5, "b": None, "c": 0.2}
+        weights = {"a": 0.2, "b": 0.3, "c": 0.1}
+        contribs, renorm = renormalize_contributions(nudges, weights)
+        assert renorm is not None and math.isclose(renorm, 2.0, rel_tol=1e-9)
+        # Вклад a = renorm·вес·нудж = 2.0·0.2·(−0.5) = −0.2.
+        assert contribs["a"] is not None
+        assert math.isclose(contribs["a"], renorm * 0.2 * -0.5, rel_tol=1e-9)
+
+    def test_not_artificially_shrunk_to_neutral(self) -> None:
+        # Один и тот же сигнал не должен ослабевать оттого, что соседние каналы
+        # деградировали: renorm возвращает «потерянный» вес оставшимся.
+        weights = {"a": 0.2, "b": 0.2, "c": 0.2}
+        # Случай 1: все доступны, все нуджи −1.
+        all_av: dict[str, float | None] = {"a": -1.0, "b": -1.0, "c": -1.0}
+        c_all, _ = renormalize_contributions(all_av, weights)
+        coef_all = assemble_coefficient(c_all)
+        # Случай 2: только a доступен (нудж −1), b/c деградировали.
+        one_av: dict[str, float | None] = {"a": -1.0, "b": None, "c": None}
+        c_one, _ = renormalize_contributions(one_av, weights)
+        coef_one = assemble_coefficient(c_one)
+        # Единственный доступный канал с тем же нуджем даёт ТОТ ЖЕ итог (renorm
+        # компенсирует выпавшие): сигнал не размазан к нейтрали.
+        assert coef_one == coef_all
+
+    def test_all_unavailable_renorm_none(self) -> None:
+        nudges: dict[str, float | None] = {"a": None, "b": None}
+        contribs, renorm = renormalize_contributions(nudges, {"a": 0.2, "b": 0.3})
+        assert renorm is None
+        assert all(v is None for v in contribs.values())
+        # Сборка из всех-None вкладов → нейтраль 1.0.
+        assert assemble_coefficient(contribs) == _MACRO_COEF_NEUTRAL
+
+    def test_available_weights_sum_back_to_total(self) -> None:
+        # Сумма ЭФФЕКТИВНЫХ весов доступных каналов (renorm·вес) = полной сумме весов
+        # (это и есть «не сжимать к 1.0»): проверяем на единичных нуджах.
+        weights = {"a": 0.18, "b": 0.12, "c": 0.10, "d": 0.20}
+        nudges: dict[str, float | None] = {"a": 1.0, "b": 1.0, "c": None, "d": None}
+        contribs, renorm = renormalize_contributions(nudges, weights)
+        eff_sum = sum(v for v in contribs.values() if v is not None)
+        assert renorm is not None
+        # eff_sum = renorm·(0.18+0.12)·1.0 = total(0.60). Доступные «забрали» c+d.
+        assert abs(eff_sum - sum(weights.values())) < 1e-12
+
+
+# ── pure: assemble_coefficient (клэмп) ────────────────────────────────────────
+
+
+class TestAssemble:
+    def test_centered_at_one(self) -> None:
+        assert assemble_coefficient({}) == _MACRO_COEF_NEUTRAL
+        assert assemble_coefficient({"a": None}) == _MACRO_COEF_NEUTRAL
+
+    def test_positive_contributions_raise(self) -> None:
+        c = assemble_coefficient({"a": 0.1, "b": 0.05})
+        assert c == _MACRO_COEF_NEUTRAL + 0.15
+
+    def test_clamped_at_min(self) -> None:
+        assert assemble_coefficient({"a": -5.0}) == _MACRO_COEF_MIN
+
+    def test_clamped_at_max(self) -> None:
+        assert assemble_coefficient({"a": 5.0}) == _MACRO_COEF_MAX
+
+    def test_none_contributions_skipped(self) -> None:
+        assert assemble_coefficient({"a": 0.1, "b": None, "c": -0.05}) == (
+            _MACRO_COEF_NEUTRAL + 0.05
+        )
+
+
+# ── compute_macro_coefficient (мок PR2) ───────────────────────────────────────
+
+
+class TestComputeMacroCoefficient:
+    def test_rate_up_issuance_down_coef_below_one(self) -> None:
+        # Ставка растёт + выдачи падают → давящий режим → coef < 1.
+        n = 12
+        months = _months(n)
+        key_rate = [10.0 + i * 0.5 for i in range(n)]  # тренд +5.5 п.п. за окно
+        issued_count = [10000.0 - i * 400 for i in range(n)]  # выдачи падают
+        issued_volume = [50000.0 - i * 2000 for i in range(n)]
+        macro = _macro(
+            months,
+            key_rate=key_rate,
+            issued_count=issued_count,
+            issued_volume=issued_volume,
+        )
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(MagicMock(), segment_profile={})
+        assert isinstance(out, MacroCoefficient)
+        assert out.coefficient < _MACRO_COEF_NEUTRAL
+        assert _F_RATE in out.available_inputs
+        assert _F_ISSUANCE in out.available_inputs
+        # Backed-каналы без данных (mortgage_rate / overdue) → unavailable, не 0.
+        assert _F_MORTG_RATE in out.unavailable_inputs
+        assert out.breakdown[_F_MORTG_RATE] is None
+
+    def test_rate_down_favored_segment_coef_above_one(self) -> None:
+        # Ставка падает + выдачи растут → поддерживающий режим → coef > 1, и для
+        # favored-сегмента (семейный/компакт) подъём КРУЧЕ (сегмент-модификатор).
+        n = 12
+        months = _months(n)
+        key_rate = [18.0 - i * 0.5 for i in range(n)]  # тренд −5.5 п.п.
+        issued_count = [8000.0 + i * 400 for i in range(n)]  # выдачи растут
+        issued_volume = [40000.0 + i * 2000 for i in range(n)]
+        macro = _macro(
+            months,
+            key_rate=key_rate,
+            issued_count=issued_count,
+            issued_volume=issued_volume,
+        )
+        with patch(_MACRO, return_value=macro):
+            neutral = compute_macro_coefficient(MagicMock(), segment_profile={})
+        with patch(_MACRO, return_value=macro):
+            favored = compute_macro_coefficient(
+                MagicMock(), segment_profile={"is_family": True, "room_bucket": "1-к 30-45"}
+            )
+        assert neutral.coefficient > _MACRO_COEF_NEUTRAL
+        assert favored.coefficient > _MACRO_COEF_NEUTRAL
+        # Favored реагирует круче на rate↓ → coef ВЫШЕ нейтрального (но клэмп может
+        # уравнять, если оба упёрлись в MAX — допускаем ≥).
+        assert favored.coefficient >= neutral.coefficient
+
+    def test_expensive_segment_steeper_negative_on_rate_up(self) -> None:
+        # Ставка растёт: дорогой/крупный сегмент должен дать coef НИЖЕ нейтрального
+        # (круче негатив на rate↑).
+        n = 12
+        months = _months(n)
+        key_rate = [9.0 + i * 0.4 for i in range(n)]
+        macro = _macro(months, key_rate=key_rate)
+        with patch(_MACRO, return_value=macro):
+            neutral = compute_macro_coefficient(MagicMock(), segment_profile={})
+        with patch(_MACRO, return_value=macro):
+            expensive = compute_macro_coefficient(
+                MagicMock(),
+                segment_profile={"obj_class": "премиум", "is_investment": True},
+            )
+        assert expensive.coefficient < neutral.coefficient
+
+    def test_all_backed_available_with_clean_window_high(self) -> None:
+        # Все 4 backed-канала есть + окно без шок-дат → confidence='high'.
+        n = 12
+        months = _months(n, end=dt.date(2023, 12, 1))  # 2023 без шок-дат PR2
+        key_rate = [9.0 + i * 0.2 for i in range(n)]
+        mortgage_rate = [11.0 + i * 0.15 for i in range(n)]
+        issued_count = [9000.0 - i * 100 for i in range(n)]
+        issued_volume = [45000.0 - i * 500 for i in range(n)]
+        debt = [2_000_000.0 + i * 1000 for i in range(n)]
+        overdue = [25_000.0 + i * 200 for i in range(n)]
+        macro = _macro(
+            months,
+            key_rate=key_rate,
+            mortgage_rate=mortgage_rate,
+            issued_count=issued_count,
+            issued_volume=issued_volume,
+            debt=debt,
+            overdue=overdue,
+        )
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(MagicMock(), segment_profile={})
+        assert out.confidence == "high"
+        assert out.confounded is False
+        # Все backed доступны; все degraded — нет.
+        for name in (_F_RATE, _F_MORTG_RATE, _F_ISSUANCE, _F_OVERDUE):
+            assert name in out.available_inputs
+        for name in _DEGRADED_FACTORS:
+            assert name in out.unavailable_inputs
+
+    def test_confounded_window_caps_confidence(self) -> None:
+        # Окно пересекает шок 2022-02 → confounded=True; даже все backed → не 'high'.
+        n = 12
+        months = _months(n, end=dt.date(2022, 6, 1))  # охватывает 2022-02
+        assert any(m == dt.date(2022, 2, 1) for m in months)
+        key_rate = [9.0 + i * 0.5 for i in range(n)]
+        mortgage_rate = [11.0 + i * 0.3 for i in range(n)]
+        issued_count = [9000.0 - i * 100 for i in range(n)]
+        issued_volume = [45000.0 - i * 500 for i in range(n)]
+        debt = [2_000_000.0 for _ in range(n)]
+        overdue = [25_000.0 + i * 300 for i in range(n)]
+        macro = _macro(
+            months,
+            key_rate=key_rate,
+            mortgage_rate=mortgage_rate,
+            issued_count=issued_count,
+            issued_volume=issued_volume,
+            debt=debt,
+            overdue=overdue,
+        )
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(MagicMock(), segment_profile={})
+        assert out.confounded is True
+        assert out.confidence != "high"
+
+    def test_partial_backed_is_medium(self) -> None:
+        # Доступны 2 backed-канала (rate + issuance) → medium (частичный сигнал).
+        n = 12
+        months = _months(n)
+        key_rate = [9.0 + i * 0.3 for i in range(n)]
+        issued_count = [9000.0 - i * 100 for i in range(n)]
+        issued_volume = [45000.0 - i * 500 for i in range(n)]
+        macro = _macro(
+            months,
+            key_rate=key_rate,
+            issued_count=issued_count,
+            issued_volume=issued_volume,
+        )
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(MagicMock(), segment_profile={})
+        assert out.confidence == "medium"
+
+    def test_graceful_empty_is_neutral_low(self) -> None:
+        # Пустой макро-ряд → coef=1.0 (нейтрально), confidence='low', не crash.
+        with patch(_MACRO, return_value=[]):
+            out = compute_macro_coefficient(MagicMock(), segment_profile={})
+        assert out.coefficient == _MACRO_COEF_NEUTRAL
+        assert out.confidence == "low"
+        assert out.confounded is False
+        assert out.weight_renorm_factor is None
+        # Все каналы недоступны → breakdown целиком None.
+        assert all(v is None for v in out.breakdown.values())
+
+    def test_all_none_series_is_neutral_low(self) -> None:
+        # Сетка месяцев есть, но все ряды None → coef=1.0, low (как degraded).
+        n = 12
+        months = _months(n)
+        macro = _macro(months)  # все ряды None
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(MagicMock(), segment_profile={})
+        assert out.coefficient == _MACRO_COEF_NEUTRAL
+        assert out.confidence == "low"
+
+    def test_none_segment_profile_defaults_neutral(self) -> None:
+        # segment_profile=None → нейтральный профиль, не crash.
+        n = 12
+        months = _months(n)
+        key_rate = [9.0 + i * 0.3 for i in range(n)]
+        macro = _macro(months, key_rate=key_rate)
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(MagicMock(), segment_profile=None)
+        assert isinstance(out, MacroCoefficient)
+        assert out.segment_profile == {}
+
+    def test_coefficient_always_within_band(self) -> None:
+        # Любой режим → coef в [MIN, MAX] (клэмп).
+        n = 12
+        months = _months(n)
+        # Жёсткий шок: ставка +6 п.п., выдачи рухнули, высокая просрочка.
+        macro = _macro(
+            months,
+            key_rate=[9.0 + i * 0.6 for i in range(n)],
+            mortgage_rate=[10.0 + i * 0.5 for i in range(n)],
+            issued_count=[10000.0 - i * 700 for i in range(n)],
+            issued_volume=[50000.0 - i * 3500 for i in range(n)],
+            debt=[1_000_000.0 for _ in range(n)],
+            overdue=[80_000.0 + i * 1000 for i in range(n)],
+        )
+        with patch(_MACRO, return_value=macro):
+            out = compute_macro_coefficient(
+                MagicMock(),
+                segment_profile={"obj_class": "премиум", "is_investment": True},
+            )
+        assert _MACRO_COEF_MIN <= out.coefficient <= _MACRO_COEF_MAX
+
+
+# ── as_dict ───────────────────────────────────────────────────────────────────
+
+
+class TestMacroCoefficientAsDict:
+    def test_serialises_and_rounds(self) -> None:
+        mc = MacroCoefficient(
+            coefficient=0.876543,
+            breakdown={"rate": -0.123456, "income": None},
+            available_inputs=["rate"],
+            unavailable_inputs=["income"],
+            segment_profile={"obj_class": "бизнес"},
+            confidence="medium",
+            confounded=False,
+            weight_renorm_factor=1.234567,
+        )
+        d = mc.as_dict()
+        assert d["coefficient"] == 0.8765
+        assert d["breakdown"]["rate"] == -0.1235
+        assert d["breakdown"]["income"] is None  # None-вклад сохраняется
+        assert d["weight_renorm_factor"] == 1.2346
+        assert d["available_inputs"] == ["rate"]
+        assert d["confidence"] == "medium"
+
+    def test_weights_schema_covers_all_factors(self) -> None:
+        # Каждый sub-factor имеет вес; degraded-набор ⊂ схеме весов.
+        assert _F_RATE in _WEIGHTS
+        assert _F_MORTG_RATE in _WEIGHTS
+        assert _F_ISSUANCE in _WEIGHTS
+        assert _F_OVERDUE in _WEIGHTS
+        assert _DEGRADED_FACTORS.issubset(set(_WEIGHTS))