backend/app/services/forecasting/__init__.py

@@ -8,7 +8,7 @@
 Слои (по PR):
   • macro_series  (#951b)          — monthly макро-ряд + классификатор режима ставки (X-ось §9.6).
   • sales_series  (#951c)          — monthly ряд продаж по сегменту (Y-ось §9.6).
-  • rate_sensitivity (later PR)    — §9.6 чувствительность продаж к key_rate.
+  • rate_sensitivity (#951d)       — §9.6 чувствительность продаж к key_rate (CORE, ADVISORY).
   • macro_coefficient (later PR)   — §9.5 макро-коэффициент.
 
 Источники данных:
@@ -25,6 +25,13 @@ from app.services.forecasting.macro_series import (
     is_confounded_window,
     macro_at_lag,
 )
+from app.services.forecasting.rate_sensitivity import (
+    RateSensitivity,
+    best_lag,
+    compute_rate_sensitivity,
+    ols_slope_r2,
+    shrink,
+)
 from app.services.forecasting.sales_series import (
     SalesSeries,
     SegmentSpec,
@@ -37,15 +44,20 @@ from app.services.forecasting.sales_series import (
 
 __all__ = [
     "MonthlyMacro",
+    "RateSensitivity",
     "SalesSeries",
     "SegmentSpec",
+    "best_lag",
     "build_sales_series",
     "classify_regime",
+    "compute_rate_sensitivity",
     "fill_month_grid",
     "get_monthly_macro",
     "is_confounded_window",
     "log_diff",
     "macro_at_lag",
+    "ols_slope_r2",
     "price_bucket_of",
     "room_area_bucket_of",
+    "shrink",
 ]

backend/app/services/forecasting/rate_sensitivity.py

@@ -0,0 +1,610 @@
+"""§9.6 чувствительность продаж сегмента к ключевой ставке ЦБ — X/Y/Z + фраза.
+
+#951 (Site Finder v2 / «GG-форсайт» ТЗ §9.6), EPIC 7 «Чувствительность к ключевой
+ставке», sub-PR 3 (#951d). Это **CORE-модуль** §9.6: для сегмента рынка регрессируем
+месячное изменение продаж на изменение ключевой ставки ЦБ при нескольких лагах,
+выбираем лучший лаг, шринкуем к EKB-широкому prior'у и собираем фразу-объяснение
+§9.6. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (numpy-математика, шаблон фразы).
+
+ADVISORY-СТАТУС: модуль остаётся СОВЕТУЮЩИМ, пока бэктест PR6 его не валидирует —
+он НЕ подключается ни к одному production-эндпоинту в этом PR. Цифры пригодны для
+explainability/прототипа, но НЕ для принятия решений до проверки на out-of-sample.
+
+Конвейер (всё на месячной сетке PR1/PR2 — окна по длине совпадают):
+  • Y-ось = Δln(units) сегмента (build_sales_series → log_diff, PR1). Стационарная,
+    рост в %.
+  • X-ось = Δ(key_rate) (get_monthly_macro, PR2). Лагируем X относительно Y, т.к.
+    реакция продаж на смену ставки запаздывает (ипотека/сделки оформляются месяцами).
+  • β = slope OLS(Δln(units) ~ Δrate[t-lag]); ЦБ ↑ставку → продажи ↓ → β<0.
+  • x_pct = 100·(exp(β)−1): % изменения продаж на +1 п.п. ставки (NEGATIVE при ↓).
+  • Z = нижняя граница площади самого чувствительного room×area-бакета (Source B).
+
+GATE-философия (зеркало analytics_queries._elasticity_coef: n≥MIN ∧ r2≥R2_MIN ∧
+slope<0 иначе FALLBACK). Адаптируем пороги под КОРОТКИЙ месячный Δ-ряд: Δln-точек
+всего ~12-48 (минус дыры/нули), а НЕ 30+ строк сделок, поэтому _MIN_OBS=8 (а не 30).
+
+SHRINKAGE: тонкий сегмент шумен → шринкуем β_seg к EKB-широкому prior'у β_EKB
+весом w=n/(n+k). Мало точек → ближе к prior; много → ≈сегмент. Это снижает дисперсию
+оценки на тонких сегментах (James-Stein дух), но именно поэтому модуль advisory до
+бэктеста: оптимальный k подбирается на out-of-sample, текущий — эвристический.
+
+Graceful-on-thin-data (дух market_metrics.py): пустой/тонкий ряд / нет прошедшего
+gate лага / выживший неправильный знак (β≥0) → confidence='low', фраза «недостаточно
+данных…», НЕ crash. Детерминированно.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass
+from datetime import date
+from typing import Any, Literal
+
+import numpy as np
+from sqlalchemy.orm import Session
+
+from app.services.forecasting.macro_series import (
+    get_monthly_macro,
+    is_confounded_window,
+)
+from app.services.forecasting.sales_series import (
+    ROOM_AREA_BUCKET_1K,
+    ROOM_AREA_BUCKET_2K,
+    ROOM_AREA_BUCKET_3K,
+    ROOM_AREA_BUCKET_LARGE,
+    ROOM_AREA_BUCKET_STUDIO,
+    SegmentSpec,
+    build_sales_series,
+    log_diff,
+)
+
+logger = logging.getLogger(__name__)
+
+Confidence = Literal["high", "medium", "low"]
+
+# ── Named-константы ───────────────────────────────────────────────────────────
+
+# Глубина ряда по умолчанию (месяцев назад). 48 мес ≈ 4 года — совпадает с
+# _DEFAULT_MONTHS_BACK PR1/PR2: §9.6 join-ит sales↔macro месяц-в-месяц, окна
+# должны быть одной длины.
+_DEFAULT_MONTHS_BACK: int = 48
+
+# Лаги (мес) X относительно Y для перебора в best_lag. 0/1/2/3/6 — те же, что
+# документированы в macro_at_lag (PR2): реакция продаж на ставку обычно
+# запаздывает на квартал, полугодовой лаг ловит долгий ипотечный эффект.
+_LAGS: tuple[int, ...] = (0, 1, 2, 3, 6)
+
+# GATE-пороги (зеркало _elasticity_coef, но адаптированы под КОРОТКИЙ месячный
+# Δ-ряд). _elasticity_coef фитит по 30+ строкам сделок; здесь же одна точка = один
+# Δln-МЕСЯЦ, а их всего ~12-48 минус дыры/нули → 30 недостижимо. 8 ≈ «минимум,
+# чтобы slope/R² не были чистым шумом на 2-3 точках» (полгода-плюс наблюдений).
+_MIN_OBS: int = 8
+# R²-минимум совпадает с _elasticity_coef (0.1): фит должен объяснять хотя бы
+# ~10% дисперсии Δln, иначе связь rate↔sales не считаем установленной.
+_MIN_R2: float = 0.1
+
+# Сила шринка к EKB-prior'у: w = n/(n+k). k=10 ≈ «сегмент с 10 Δln-точками весит
+# поровну с prior'ом (w=0.5)»; на ~24 точках w≈0.7 (доминирует сегмент), на тонких
+# ~6 точках w≈0.375 (тянет к prior'у). Подобрано под глубину месячного ряда (≤48);
+# advisory — точный k валидирует бэктест PR6.
+_SHRINK_K: float = 10.0
+
+# Confidence-пороги (зеркало духа PR1/PR2). 'high' требует И длины ряда, И силы
+# фита, И доминирования сегмента над prior'ом, И отсутствия шок-окна.
+_CONF_HIGH_MIN_OBS: int = 24  # ≥2 года Δln-наблюдений
+_CONF_HIGH_MIN_R2: float = 0.3  # фит объясняет ≥30% дисперсии
+_CONF_HIGH_MIN_WEIGHT: float = 0.7  # сегмент доминирует над prior'ом
+
+# Порог shrinkage-веса, ниже которого per-segment β считаем недостаточно
+# самостоятельным для адресной фразы → деградируем во 2-м предложении к
+# EKB-широкой форме (без «наиболее чувствительны …»).
+_PHRASE_SEGMENT_WEIGHT_MIN: float = 0.5
+
+# Нижняя граница площади (м²) для room×area-бакетов Source B (Z-выход §9.6).
+# Зеркало area-меток ROOM_AREA_BUCKET_* (sales_series): "Студии 15-30"→15,
+# "1-к 30-45"→30, "2-к 45-60"→45, "3-к 60-80"→60, "80+ м²"→80. Держим явной
+# таблицей (а не парсингом метки), чтобы переименование метки не сломало Z молча.
+_BUCKET_AREA_FLOOR_M2: dict[str, float] = {
+    ROOM_AREA_BUCKET_STUDIO: 15.0,
+    ROOM_AREA_BUCKET_1K: 30.0,
+    ROOM_AREA_BUCKET_2K: 45.0,
+    ROOM_AREA_BUCKET_3K: 60.0,
+    ROOM_AREA_BUCKET_LARGE: 80.0,
+}
+
+# Source B room-бакеты, по которым ищем «самый чувствительный» (Z-выход). Только
+# area-меченые — unknown-бакет осмысленной площади не несёт, в Z-перебор не идёт.
+_SOURCE_B_ROOM_BUCKETS: tuple[str, ...] = (
+    ROOM_AREA_BUCKET_STUDIO,
+    ROOM_AREA_BUCKET_1K,
+    ROOM_AREA_BUCKET_2K,
+    ROOM_AREA_BUCKET_3K,
+    ROOM_AREA_BUCKET_LARGE,
+)
+
+# Шаблоны фразы §9.6 (НЕ LLM). Полная форма с адресным бакетом / EKB-широкая форма
+# (когда per-bucket слаб) / форма «нет данных» (нет gate-лага или выжил β≥0).
+_PHRASE_FULL: str = (
+    "При росте ключевой ставки на 1 п.п. в аналогичных условиях продажи в этой "
+    "локации снижались в среднем на {x}% через {y} месяцев. Наиболее чувствительны "
+    "{bucket} площадью более {z} м²."
+)
+_PHRASE_EKB_WIDE: str = (
+    "При росте ключевой ставки на 1 п.п. в аналогичных условиях продажи в этой "
+    "локации снижались в среднем на {x}% через {y} месяцев."
+)
+_PHRASE_INSUFFICIENT: str = "недостаточно данных для оценки чувствительности к ставке"
+
+_SOURCE_A: Literal["corpus_room_month"] = "corpus_room_month"
+_SOURCE_B: Literal["objective_lots"] = "objective_lots"
+
+
+@dataclass(frozen=True)
+class RateSensitivity:
+    """§9.6 чувствительность продаж сегмента к ключевой ставке ЦБ.
+
+    Детерминированный результат. Любое числовое поле = None при недостатке данных
+    (никогда 0-как-заглушка). `phrase` ВСЕГДА заполнена (в т.ч. «недостаточно
+    данных…»). ADVISORY до валидации бэктестом PR6 — не для production-решений.
+    """
+
+    segment: dict[str, str | None]
+    x_pct: float | None  # % изменения продаж на +1 п.п. ставки (NEGATIVE при ↓)
+    y_lag_months: int | None  # лаг (мес), на котором связь сильнейшая
+    z_area_floor: float | None  # нижняя граница площади самого чувствит. бакета, м²
+    most_sensitive_bucket: str | None  # метка этого бакета (Source B)
+    beta: float | None  # шринкнутый slope на Δln (сырой коэффициент)
+    r2: float | None  # R² выигравшего лага сегмента
+    n_obs: int  # число Δln-точек, по которым фитили сегмент
+    shrinkage_weight: float  # w = n/(n+k) ∈ [0,1] (вес сегмента против prior'а)
+    confounded: bool  # окно ряда пересекает шок-дату (PR2)
+    confidence: Confidence
+    phrase: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "segment": dict(self.segment),
+            "x_pct": _round_or_none(self.x_pct, 1),
+            "y_lag_months": self.y_lag_months,
+            "z_area_floor": _round_or_none(self.z_area_floor, 1),
+            "most_sensitive_bucket": self.most_sensitive_bucket,
+            "beta": _round_or_none(self.beta, 4),
+            "r2": _round_or_none(self.r2, 4),
+            "n_obs": self.n_obs,
+            "shrinkage_weight": round(self.shrinkage_weight, 3),
+            "confounded": self.confounded,
+            "confidence": self.confidence,
+            "phrase": self.phrase,
+        }
+
+
+def _round_or_none(value: float | None, digits: int) -> float | None:
+    return round(value, digits) if value is not None else None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pure-логика — без БД, полностью юнит-тестируемо (numpy-математика на синтетике).
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def ols_slope_r2(
+    x: list[float | None], y: list[float | None]
+) -> tuple[float | None, float | None, int]:
+    """Univariate OLS y~x через numpy → (slope, R², n) на парных конечных точках.
+
+    Дропаем любую пару, где x или y = None / NaN / Inf (Δln-ряды дырявые: нули и
+    первая точка → None). Slope/R² через numpy.polyfit(deg=1) + ручной R²
+    (1 − SS_res/SS_tot). PURE, без БД.
+
+    Возвращает (None, None, n) если n<2 (нечего фитить) или нулевая дисперсия x/y
+    (вертикальная/горизонтальная линия — slope/R² не определены).
+
+    Args:
+        x: регрессор (обычно Δrate при лаге), None/NaN допустимы.
+        y: зависимая (обычно Δln(units)), None/NaN допустимы.
+
+    Returns:
+        (slope, r2, n_used): slope/r2 = None при невозможности фита; n_used —
+        число использованных (конечных) пар.
+    """
+    xs: list[float] = []
+    ys: list[float] = []
+    for xv, yv in zip(x, y, strict=False):
+        if xv is None or yv is None:
+            continue
+        xf = float(xv)
+        yf = float(yv)
+        if not (math.isfinite(xf) and math.isfinite(yf)):
+            continue
+        xs.append(xf)
+        ys.append(yf)
+
+    n = len(xs)
+    if n < 2:
+        return None, None, n
+
+    xa = np.asarray(xs, dtype=float)
+    ya = np.asarray(ys, dtype=float)
+    # Нулевая дисперсия → slope/R² не определены (деление на ноль в фите/R²).
+    if float(np.var(xa)) == 0.0 or float(np.var(ya)) == 0.0:
+        return None, None, n
+
+    slope, intercept = np.polyfit(xa, ya, 1)
+    y_hat = slope * xa + intercept
+    ss_res = float(np.sum((ya - y_hat) ** 2))
+    ss_tot = float(np.sum((ya - np.mean(ya)) ** 2))
+    if ss_tot == 0.0:  # уже отсечено var(y)==0, но страховка от FP-краёв
+        return None, None, n
+    r2 = 1.0 - ss_res / ss_tot
+    return float(slope), float(r2), n
+
+
+def _delta(series: list[float | None]) -> list[float | None]:
+    """Простая первая разность ряда: out[t] = x_t − x_{t-1}; out[0]=None.
+
+    Применяем к key_rate (Δ в п.п.). None, если любая из двух точек None. PURE.
+    """
+    out: list[float | None] = [None]
+    for i in range(1, len(series)):
+        cur = series[i]
+        prev = series[i - 1]
+        if cur is None or prev is None:
+            out.append(None)
+        else:
+            out.append(float(cur) - float(prev))
+    return out
+
+
+def best_lag(
+    deltas_sales: list[float | None],
+    monthly_macro_rate_deltas: list[float | None],
+    *,
+    lags: tuple[int, ...] = _LAGS,
+    min_obs: int = _MIN_OBS,
+    min_r2: float = _MIN_R2,
+) -> dict[str, Any] | None:
+    """Выбрать лаг с сильнейшей (самой негативной) gate-прошедшей связью rate→sales.
+
+    Для каждого лага L выравниваем Δln(sales)[t] с Δrate[t−L] (rate ОТСТАЁТ: продажи
+    реагируют на прошлое изменение ставки), фитим ols_slope_r2, применяем GATE
+    (n≥min_obs ∧ r2≥min_r2 ∧ slope<0 — зеркало _elasticity_coef). Среди прошедших
+    берём с НАИБОЛЕЕ НЕГАТИВНЫМ slope (сильнейшее падение продаж). Оба ряда должны
+    быть одной длины (общая месячная сетка PR1/PR2). PURE, без БД.
+
+    Args:
+        deltas_sales: Δln(units) по месяцам (из log_diff, PR1), None-дыры ок.
+        monthly_macro_rate_deltas: Δ(key_rate) по тем же месяцам (None-дыры ок).
+        lags: лаги-кандидаты (по умолчанию _LAGS).
+        min_obs: GATE — минимум парных точек.
+        min_r2: GATE — минимум R².
+
+    Returns:
+        {'lag','slope','r2','n'} лучшего прошедшего лага, либо None если ни один не
+        прошёл gate (тонко / неправильный знак / слабый фит).
+    """
+    best: dict[str, Any] | None = None
+    n_months = len(deltas_sales)
+    for lag in lags:
+        # X лагируется: y[t] ← x[t-lag]. Сдвигаем X вправо на lag (первые lag = None).
+        shifted_x: list[float | None] = [None] * lag + list(monthly_macro_rate_deltas)
+        shifted_x = shifted_x[:n_months]
+        slope, r2, n = ols_slope_r2(shifted_x, deltas_sales)
+        if slope is None or r2 is None:
+            continue
+        if n < min_obs or r2 < min_r2 or slope >= 0:
+            continue
+        if best is None or slope < best["slope"]:
+            best = {"lag": lag, "slope": slope, "r2": r2, "n": n}
+    return best
+
+
+def shrink(
+    beta_seg: float, n_seg: int, beta_ekb: float, *, k: float = _SHRINK_K
+) -> tuple[float, float]:
+    """Шринк сегментного β к EKB-prior'у: w=n/(n+k); β=w·β_seg+(1−w)·β_ekb.
+
+    Тонкий сегмент (мало n) шумен → тянем к более стабильному EKB-широкому
+    prior'у. Много n → w→1 → ≈β_seg. PURE, без БД.
+
+    Args:
+        beta_seg: сырой slope сегмента (Δln).
+        n_seg: число Δln-точек сегмента (вес доверия).
+        beta_ekb: EKB-широкий prior-slope.
+        k: сила шринка (по умолчанию _SHRINK_K).
+
+    Returns:
+        (beta_shrunk, weight): шринкнутый slope и вес w∈[0,1] сегмента.
+    """
+    n = max(0, n_seg)
+    denom = n + k
+    w = (n / denom) if denom > 0 else 0.0
+    shrunk = w * beta_seg + (1.0 - w) * beta_ekb
+    return shrunk, w
+
+
+def _x_pct_from_beta(beta: float) -> float:
+    """β на Δln → % изменения продаж на +1 п.п. ставки: 100·(exp(β)−1).
+
+    β<0 → отрицательный % (продажи падают). exp-преобразование т.к. Y = Δln
+    (мультипликативный/процентный масштаб). PURE.
+    """
+    return 100.0 * (math.exp(beta) - 1.0)
+
+
+def _confidence(
+    *, n_obs: int, r2: float | None, weight: float, confounded: bool, gated: bool
+) -> Confidence:
+    """Confidence §9.6 (см. _CONF_HIGH_* / gate). PURE.
+
+    'high'   — длинный ряд И сильный фит И доминирование сегмента И не шок-окно.
+    'medium' — прошёл gate (есть валидный негативный лаг), но не дотянул до high.
+    'low'    — gate не прошёл (тонко / неправильный знак / слабо).
+    """
+    if not gated:
+        return "low"
+    if (
+        n_obs >= _CONF_HIGH_MIN_OBS
+        and r2 is not None
+        and r2 >= _CONF_HIGH_MIN_R2
+        and weight >= _CONF_HIGH_MIN_WEIGHT
+        and not confounded
+    ):
+        return "high"
+    return "medium"
+
+
+def _build_phrase(
+    *,
+    beta: float | None,
+    x_pct: float | None,
+    y_lag_months: int | None,
+    weight: float,
+    most_sensitive_bucket: str | None,
+    z_area_floor: float | None,
+) -> str:
+    """Собрать фразу §9.6 по шаблону (НЕ LLM) с правилами деградации. PURE.
+
+    Деградация:
+      • нет gate-лага (beta/x_pct/lag = None) ИЛИ выживший неправильный знак
+        (beta≥0 → продажи якобы РАСТУТ при ↑ставки — не доверяем) → «недостаточно
+        данных…».
+      • есть валидный негативный β, но per-segment слаб (weight < порога) ИЛИ
+        нет адресного бакета/Z → EKB-широкая форма (1-е предложение).
+      • иначе → полная форма с «наиболее чувствительны {bucket} … {z} м²».
+    X в фразе — положительная МАГНИТУДА («снижались на X%»).
+    """
+    if beta is None or x_pct is None or y_lag_months is None or beta >= 0:
+        return _PHRASE_INSUFFICIENT
+    x_mag = round(abs(x_pct), 1)
+    if weight < _PHRASE_SEGMENT_WEIGHT_MIN or most_sensitive_bucket is None or z_area_floor is None:
+        return _PHRASE_EKB_WIDE.format(x=x_mag, y=y_lag_months)
+    return _PHRASE_FULL.format(
+        x=x_mag,
+        y=y_lag_months,
+        bucket=most_sensitive_bucket,
+        z=round(z_area_floor, 1),
+    )
+
+
+def _insufficient(
+    *, segment: dict[str, str | None], n_obs: int, weight: float, confounded: bool
+) -> RateSensitivity:
+    """Граничный результат «недостаточно данных» (low, фраза-заглушка). PURE."""
+    return RateSensitivity(
+        segment=segment,
+        x_pct=None,
+        y_lag_months=None,
+        z_area_floor=None,
+        most_sensitive_bucket=None,
+        beta=None,
+        r2=None,
+        n_obs=n_obs,
+        shrinkage_weight=weight,
+        confounded=confounded,
+        confidence="low",
+        phrase=_PHRASE_INSUFFICIENT,
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# DB-оркестратор — тонкий, graceful. Pure-логика выше тестируется без него.
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def compute_rate_sensitivity(
+    db: Session,
+    *,
+    spec: SegmentSpec,
+    months_back: int = _DEFAULT_MONTHS_BACK,
+) -> RateSensitivity:
+    """Вычислить §9.6 чувствительность сегмента к ключевой ставке (X/Y/Z + фраза).
+
+    ADVISORY до валидации бэктестом PR6 — НЕ подключать в production-эндпоинт.
+
+    Шаги:
+      1. EKB-широкий prior: ряд продаж spec-all-None (Source A) + макро → β_EKB
+         через best_lag (bootstrap-prior для шринка; если EKB-фит сам не прошёл
+         gate → prior=0.0, т.е. шринк тянет тонкий сегмент к «нет эффекта»).
+      2. Целевой сегмент: ряд продаж по spec (Source A) → β_seg → шринк к β_EKB.
+      3. Z: перебираем Source-B room×area-бакеты, берём самый чувствительный
+         (самый негативный gate-прошедший β); Z = нижняя граница его площади.
+      4. x_pct=100·(exp(β_shrunk)−1); y_lag=выигравший лаг; confounded по окну ряда.
+      5. confidence + фраза (с деградацией: неправильный знак / тонко → insufficient;
+         слабый per-segment → EKB-широкая форма).
+
+    Graceful: тонко/пусто/ошибка → confidence='low', фраза «недостаточно данных…»,
+    НЕ crash. Детерминированно.
+
+    Args:
+        db: SQLAlchemy sync Session.
+        spec: целевой сегмент (любой subset осей).
+        months_back: глубина ряда (по умолчанию _DEFAULT_MONTHS_BACK).
+
+    Returns:
+        RateSensitivity (всегда; фраза заполнена даже при нехватке данных).
+    """
+    segment = spec.as_dict()
+
+    macro = get_monthly_macro(db, months_back=months_back)
+    rate_deltas = _delta([m.key_rate for m in macro])
+    macro_months = [m.month for m in macro]
+
+    # ── 1. EKB-широкий prior (bootstrap для шринка) ────────────────────────────
+    ekb_spec = SegmentSpec()  # all-None → агрегат по всему ЕКБ
+    ekb_sales = build_sales_series(db, spec=ekb_spec, source=_SOURCE_A, months_back=months_back)
+    ekb_deltas = _align_sales_deltas(ekb_sales.months, ekb_sales.units, macro_months)
+    ekb_fit = best_lag(ekb_deltas, rate_deltas)
+    # Если EKB-фит сам не прошёл gate — prior = нейтральный 0.0 (нет эффекта): тонкий
+    # сегмент тогда шринкуется к «ставка на продажи не влияет», а не к шуму.
+    beta_ekb = float(ekb_fit["slope"]) if ekb_fit is not None else 0.0
+
+    # ── 2. Целевой сегмент → β_seg → шринк ─────────────────────────────────────
+    seg_sales = build_sales_series(db, spec=spec, source=_SOURCE_A, months_back=months_back)
+    seg_deltas = _align_sales_deltas(seg_sales.months, seg_sales.units, macro_months)
+    seg_fit = best_lag(seg_deltas, rate_deltas)
+
+    confounded = _series_confounded(seg_sales.months)
+
+    if seg_fit is None:
+        # Сегмент сам не дал валидного негативного лага → недостаточно данных.
+        _, w0 = shrink(0.0, 0, beta_ekb)
+        logger.info(
+            "rate_sensitivity: no gated lag (segment=%s n_months=%d) → insufficient",
+            segment,
+            seg_sales.n_months,
+        )
+        return _insufficient(segment=segment, n_obs=0, weight=w0, confounded=confounded)
+
+    beta_seg = float(seg_fit["slope"])
+    n_seg = int(seg_fit["n"])
+    r2_seg = float(seg_fit["r2"])
+    y_lag = int(seg_fit["lag"])
+    beta_shrunk, weight = shrink(beta_seg, n_seg, beta_ekb)
+
+    # Шринк к prior'у может перевернуть знак (β_seg<0, но β_ekb>0 и доминирует) →
+    # эффект уже не «продажи падают». Тогда деградируем в insufficient (см. _build_phrase).
+    if beta_shrunk >= 0:
+        logger.info(
+            "rate_sensitivity: shrunk beta non-negative (segment=%s beta_seg=%.4f "
+            "beta_ekb=%.4f w=%.3f) → insufficient",
+            segment,
+            beta_seg,
+            beta_ekb,
+            weight,
+        )
+        return _insufficient(segment=segment, n_obs=n_seg, weight=weight, confounded=confounded)
+
+    # ── 3. Z: самый чувствительный Source-B room×area-бакет ────────────────────
+    most_sensitive_bucket, z_area_floor = _most_sensitive_bucket(
+        db, spec=spec, rate_deltas=rate_deltas, macro_months=macro_months, months_back=months_back
+    )
+
+    # ── 4. X / lag / confounded ────────────────────────────────────────────────
+    x_pct = _x_pct_from_beta(beta_shrunk)
+
+    # ── 5. confidence + фраза ──────────────────────────────────────────────────
+    confidence = _confidence(
+        n_obs=n_seg, r2=r2_seg, weight=weight, confounded=confounded, gated=True
+    )
+    phrase = _build_phrase(
+        beta=beta_shrunk,
+        x_pct=x_pct,
+        y_lag_months=y_lag,
+        weight=weight,
+        most_sensitive_bucket=most_sensitive_bucket,
+        z_area_floor=z_area_floor,
+    )
+
+    logger.info(
+        "rate_sensitivity: segment=%s beta_shrunk=%.4f x_pct=%.1f lag=%d w=%.3f "
+        "bucket=%s z=%s confounded=%s confidence=%s",
+        segment,
+        beta_shrunk,
+        x_pct,
+        y_lag,
+        weight,
+        most_sensitive_bucket,
+        z_area_floor,
+        confounded,
+        confidence,
+    )
+
+    return RateSensitivity(
+        segment=segment,
+        x_pct=x_pct,
+        y_lag_months=y_lag,
+        z_area_floor=z_area_floor,
+        most_sensitive_bucket=most_sensitive_bucket,
+        beta=beta_shrunk,
+        r2=r2_seg,
+        n_obs=n_seg,
+        shrinkage_weight=weight,
+        confounded=confounded,
+        confidence=confidence,
+        phrase=phrase,
+    )
+
+
+def _align_sales_deltas(
+    sales_months: list[date], sales_units: list[int], macro_months: list[date]
+) -> list[float | None]:
+    """Выровнять Δln(units) сегмента по сетке макро-месяцев (общая ось X↔Y).
+
+    log_diff даёт Δln по сетке ПРОДАЖ; чтобы пары (Δrate[t−L], Δln[t]) были month-
+    в-month, перекладываем их на сетку macro_months (месяц без продаж в выравнивании
+    → None). Так best_lag сдвигает X строго по той же временной оси. PURE.
+    """
+    deltas = log_diff(sales_units)
+    by_month = dict(zip(sales_months, deltas, strict=False))
+    return [by_month.get(m) for m in macro_months]
+
+
+def _series_confounded(months: list[date]) -> bool:
+    """True, если окно ряда [min..max] пересекает шок-дату (reuse PR2). Пустой → False."""
+    if not months:
+        return False
+    return is_confounded_window(min(months), max(months))
+
+
+def _most_sensitive_bucket(
+    db: Session,
+    *,
+    spec: SegmentSpec,
+    rate_deltas: list[float | None],
+    macro_months: list[date],
+    months_back: int,
+) -> tuple[str | None, float | None]:
+    """Найти самый чувствительный Source-B room×area-бакет → (метка, area_floor).
+
+    Перебираем area-меченые бакеты (Source B несёт area/price, Source A — нет),
+    для каждого строим ряд (тот же class/district, room_bucket=метка), фитим best_lag
+    и берём бакет с самым НЕГАТИВНЫМ gate-прошедшим slope. Z = нижняя граница его
+    площади (_BUCKET_AREA_FLOOR_M2). Ни один бакет не прошёл gate → (None, None) —
+    фраза деградирует к EKB-широкой форме.
+
+    Graceful: build_sales_series сам не бросает на тонких данных; здесь не крашим.
+    """
+    best_bucket: str | None = None
+    best_slope: float | None = None
+    for bucket in _SOURCE_B_ROOM_BUCKETS:
+        bucket_spec = SegmentSpec(
+            obj_class=spec.obj_class,
+            room_bucket=bucket,
+            district=spec.district,
+            price_bucket=spec.price_bucket,
+        )
+        bucket_sales = build_sales_series(
+            db, spec=bucket_spec, source=_SOURCE_B, months_back=months_back
+        )
+        bucket_deltas = _align_sales_deltas(bucket_sales.months, bucket_sales.units, macro_months)
+        fit = best_lag(bucket_deltas, rate_deltas)
+        if fit is None:
+            continue
+        slope = float(fit["slope"])
+        if best_slope is None or slope < best_slope:
+            best_slope = slope
+            best_bucket = bucket
+    if best_bucket is None:
+        return None, None
+    return best_bucket, _BUCKET_AREA_FLOOR_M2.get(best_bucket)

backend/tests/services/forecasting/test_rate_sensitivity.py

@@ -0,0 +1,513 @@
+"""Unit-тесты §9.6 чувствительности к ключевой ставке (#951d, CORE-модуль).
+
+Чистые тесты — БЕЗ живой БД (numpy-математика на синтетике + мок PR1/PR2):
+  • ols_slope_r2 — восстановление известного slope из y=a+b·x+шум; нулевая
+    дисперсия → None; <2 точек → None; дроп None/NaN-пар.
+  • best_lag — ловит инжектированный лаг (sales реагируют на rate на лаге 3);
+    отвергает неправильный знак; None когда ни один лаг не прошёл gate.
+  • shrink — математика w=n/(n+k); тонкий n тянет к prior'у; большой n ≈ сегмент.
+  • compute_rate_sensitivity (мок build_sales_series + get_monthly_macro): фраза
+    заполняет X/Y/Z; деградирует (insufficient, EKB-широкая форма) на тонком/
+    неправильном знаке; confounded-флаг протекает; graceful пусто → low; знак x_pct.
+
+ADVISORY-статус (до бэктеста PR6) проверяется на уровне поведения деградации.
+"""
+
+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_series import MonthlyMacro
+from app.services.forecasting.rate_sensitivity import (
+    _BUCKET_AREA_FLOOR_M2,
+    _MIN_OBS,
+    _PHRASE_INSUFFICIENT,
+    _SHRINK_K,
+    RateSensitivity,
+    _delta,
+    best_lag,
+    compute_rate_sensitivity,
+    ols_slope_r2,
+    shrink,
+)
+from app.services.forecasting.sales_series import (
+    ROOM_AREA_BUCKET_2K,
+    SalesSeries,
+    SegmentSpec,
+)
+
+_BUILD = "app.services.forecasting.rate_sensitivity.build_sales_series"
+_MACRO = "app.services.forecasting.rate_sensitivity.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))
+
+
+# ── pure: ols_slope_r2 ────────────────────────────────────────────────────────
+
+
+class TestOlsSlopeR2:
+    def test_recovers_known_slope(self) -> None:
+        # y = 2 + (-1.5)·x + крошечный детерминированный «шум» → slope ≈ -1.5, R² высок.
+        xs = [float(i) for i in range(20)]
+        noise = [((i % 3) - 1) * 0.001 for i in range(20)]
+        ys = [2.0 - 1.5 * x + n for x, n in zip(xs, noise, strict=True)]
+        slope, r2, n = ols_slope_r2(xs, ys)
+        assert n == 20
+        assert slope is not None and r2 is not None
+        assert math.isclose(slope, -1.5, abs_tol=1e-3)
+        assert r2 > 0.999
+
+    def test_recovers_positive_slope(self) -> None:
+        xs = [float(i) for i in range(15)]
+        ys = [1.0 + 0.8 * x for x in xs]
+        slope, r2, n = ols_slope_r2(xs, ys)
+        assert slope is not None and math.isclose(slope, 0.8, abs_tol=1e-6)
+        assert r2 is not None and math.isclose(r2, 1.0, abs_tol=1e-9)
+        assert n == 15
+
+    def test_zero_variance_x_returns_none(self) -> None:
+        slope, r2, n = ols_slope_r2([3.0, 3.0, 3.0], [1.0, 2.0, 3.0])
+        assert slope is None and r2 is None and n == 3
+
+    def test_zero_variance_y_returns_none(self) -> None:
+        slope, r2, n = ols_slope_r2([1.0, 2.0, 3.0], [5.0, 5.0, 5.0])
+        assert slope is None and r2 is None and n == 3
+
+    def test_fewer_than_two_points_returns_none(self) -> None:
+        assert ols_slope_r2([1.0], [2.0]) == (None, None, 1)
+        assert ols_slope_r2([], []) == (None, None, 0)
+
+    def test_drops_none_pairs(self) -> None:
+        # None в любой позиции пары → пара выбрасывается; slope считается по остатку.
+        xs: list[float | None] = [0.0, None, 2.0, 3.0, None]
+        ys: list[float | None] = [0.0, 5.0, -2.0, -3.0, 9.0]
+        slope, _r2, n = ols_slope_r2(xs, ys)
+        assert n == 3  # (0,0),(2,-2),(3,-3)
+        assert slope is not None and math.isclose(slope, -1.0, abs_tol=1e-9)
+
+    def test_drops_nan_and_inf_pairs(self) -> None:
+        xs = [0.0, 1.0, 2.0, float("nan"), float("inf")]
+        ys = [0.0, -1.0, -2.0, 10.0, 10.0]
+        slope, _r2, n = ols_slope_r2(xs, ys)
+        assert n == 3
+        assert slope is not None and math.isclose(slope, -1.0, abs_tol=1e-9)
+
+
+# ── pure: _delta ──────────────────────────────────────────────────────────────
+
+
+class TestDelta:
+    def test_first_diff(self) -> None:
+        assert _delta([10.0, 12.0, 11.0]) == [None, 2.0, -1.0]
+
+    def test_none_breaks_pair(self) -> None:
+        assert _delta([1.0, None, 3.0]) == [None, None, None]
+
+    def test_empty_and_single(self) -> None:
+        assert _delta([]) == [None]  # out always starts with None sentinel
+        assert _delta([5.0]) == [None]
+
+
+# ── pure: best_lag ────────────────────────────────────────────────────────────
+
+
+class TestBestLag:
+    def _rate_deltas(self, n: int) -> list[float | None]:
+        """Детерминированный Δrate с НИЗКОЙ автокорреляцией.
+
+        Важно: периодический (sin) регрессор имеет знак-переворачивающую
+        автокорреляцию (corr(x[t], x[t-3]) < 0 для sin), поэтому позитивная связь
+        на лаге 0 ложно «всплывает» негативной на лаге 3. Берём апериодический
+        зубчатый ряд (LCG-стиль, центрированный) — лаги слабо коррелируют, так что
+        выигрывает ИСТИННЫЙ инжектированный лаг, а «неправильный знак» остаётся
+        неправильным на всех лагах.
+        """
+        out: list[float | None] = []
+        state = 7
+        for _ in range(n):
+            state = (state * 1103515245 + 12345) % 2147483648
+            out.append(state / 2147483648.0 - 0.5)  # ∈ [-0.5, 0.5)
+        return out
+
+    def test_picks_injected_lag_three(self) -> None:
+        # sales[t] реагируют на rate[t-3]: Δln(sales)[t] = -2.0·Δrate[t-3].
+        n = 36
+        rate = self._rate_deltas(n)
+        lag = 3
+        b = -2.0
+        sales: list[float | None] = []
+        for t in range(n):
+            src = rate[t - lag] if t - lag >= 0 else None
+            sales.append(b * src if src is not None else None)
+        out = best_lag(sales, rate)
+        assert out is not None
+        assert out["lag"] == 3
+        assert math.isclose(out["slope"], b, abs_tol=1e-6)
+        assert out["r2"] > 0.99
+        assert out["n"] >= _MIN_OBS
+
+    def test_rejects_wrong_sign(self) -> None:
+        # Положительная связь (продажи РАСТУТ при ↑ставки): на апериодическом
+        # регрессоре slope позитивен на всех лагах → gate(slope<0) отвергает все.
+        n = 30
+        rate = self._rate_deltas(n)
+        sales: list[float | None] = [1.5 * r for r in rate]  # b>0
+        assert best_lag(sales, rate) is None
+
+    def test_none_when_too_few_obs(self) -> None:
+        # Идеальная негативная связь, но всего < _MIN_OBS точек → gate(n) не пускает.
+        n = 5
+        rate = self._rate_deltas(n)
+        sales: list[float | None] = [-2.0 * r for r in rate]
+        assert best_lag(sales, rate) is None
+
+    def test_none_when_no_correlation(self) -> None:
+        # Y — нулевая дисперсия после дропа: фит невозможен → None.
+        n = 30
+        rate = self._rate_deltas(n)
+        sales: list[float | None] = [0.0 for _ in range(n)]
+        assert best_lag(sales, rate) is None
+
+    def test_prefers_most_negative_slope(self) -> None:
+        # Чистый лаг-0 vs лаг-1 на ОРТОГОНАЛЬНЫХ драйверах: лаг 0 → b=-1, лаг 1 → b=-3.
+        # Используем разреженные импульсы на чётных/нечётных t, чтобы лаги не смешивались.
+        n = 40
+        rate = self._rate_deltas(n)
+        sales: list[float | None] = []
+        for t in range(n):
+            v0 = -1.0 * rate[t]
+            v1 = -3.0 * rate[t - 1] if t - 1 >= 0 else 0.0
+            sales.append(v0 + v1)
+        out = best_lag(sales, rate, lags=(0, 1))
+        assert out is not None and out["lag"] == 1  # -3 негативнее -1
+
+
+# ── pure: shrink ──────────────────────────────────────────────────────────────
+
+
+class TestShrink:
+    def test_weight_math(self) -> None:
+        shrunk, w = shrink(beta_seg=-2.0, n_seg=10, beta_ekb=-1.0, k=10.0)
+        assert math.isclose(w, 10 / 20)  # n/(n+k)
+        assert math.isclose(shrunk, 0.5 * -2.0 + 0.5 * -1.0)
+
+    def test_thin_segment_pulled_toward_prior(self) -> None:
+        # n=2, k=10 → w≈0.167 → шринкнутый ближе к prior'у (-1.0), чем к сегменту (-5.0).
+        shrunk, w = shrink(beta_seg=-5.0, n_seg=2, beta_ekb=-1.0, k=_SHRINK_K)
+        assert w < 0.3
+        assert abs(shrunk - (-1.0)) < abs(shrunk - (-5.0))
+
+    def test_large_segment_approx_segment(self) -> None:
+        # n=1000, k=10 → w≈0.99 → шринкнутый ≈ сегмент.
+        shrunk, w = shrink(beta_seg=-3.0, n_seg=1000, beta_ekb=0.0, k=_SHRINK_K)
+        assert w > 0.98
+        assert math.isclose(shrunk, -3.0, abs_tol=0.05)
+
+    def test_zero_n_is_full_prior(self) -> None:
+        shrunk, w = shrink(beta_seg=-9.0, n_seg=0, beta_ekb=-1.0, k=_SHRINK_K)
+        assert w == 0.0
+        assert math.isclose(shrunk, -1.0)
+
+
+# ── compute_rate_sensitivity (мок PR1/PR2) ────────────────────────────────────
+
+
+def _macro_series(rate_levels: list[float | None], months: list[dt.date]) -> list[MonthlyMacro]:
+    """Список MonthlyMacro с заданными уровнями key_rate (прочие поля None)."""
+    out: list[MonthlyMacro] = []
+    for month, kr in zip(months, rate_levels, strict=True):
+        out.append(
+            MonthlyMacro(
+                month=month,
+                key_rate=kr,
+                mortgage_rate_weighted=None,
+                mortgage_issued_count=None,
+                mortgage_issued_volume=None,
+                mortgage_debt=None,
+                mortgage_overdue=None,
+            )
+        )
+    return out
+
+
+def _sales(months: list[dt.date], units: list[int], *, source: str) -> SalesSeries:
+    return SalesSeries(
+        months=months,
+        units=units,
+        area_m2=[None] * len(months),
+        avg_price_per_m2=[None] * len(months),
+        n_months=len(months),
+        source=source,  # type: ignore[arg-type]
+        segment={},
+        confidence="high",
+    )
+
+
+def _synth_sales_units(
+    rate_levels: list[float],
+    *,
+    lag: int,
+    beta: float,
+    base: float = 1000.0,
+) -> list[int]:
+    """units так, что log_diff(units)[t] ≈ beta·Δrate[t-lag] (инжектируем связь).
+
+    Строим уровни мультипликативно: ln(u_t) = ln(u_{t-1}) + beta·Δrate[t-lag].
+    Округляем в int (units — счётчик); шаг малый, чтобы округление не убило связь.
+    """
+    rate_deltas = [0.0] + [rate_levels[i] - rate_levels[i - 1] for i in range(1, len(rate_levels))]
+    ln_u = math.log(base)
+    units: list[int] = [round(math.exp(ln_u))]
+    for t in range(1, len(rate_levels)):
+        src = rate_deltas[t - lag] if t - lag >= 0 else 0.0
+        ln_u += beta * src
+        units.append(max(1, round(math.exp(ln_u))))
+    return units
+
+
+class _Dispatcher:
+    """side_effect для build_sales_series: разный ряд per (source, room_bucket)."""
+
+    def __init__(
+        self,
+        *,
+        months: list[dt.date],
+        source_a_units: list[int],
+        bucket_units: dict[str, list[int]] | None = None,
+        default_units: list[int] | None = None,
+    ) -> None:
+        self.months = months
+        self.source_a_units = source_a_units
+        self.bucket_units = bucket_units or {}
+        self.default_units = default_units or [0] * len(months)
+
+    def __call__(
+        self, _db: object, *, spec: SegmentSpec, source: str, **_kw: object
+    ) -> SalesSeries:
+        if source == "corpus_room_month":
+            return _sales(self.months, self.source_a_units, source=source)
+        # Source B — per-bucket.
+        units = self.bucket_units.get(spec.room_bucket or "", self.default_units)
+        return _sales(self.months, units, source=source)
+
+
+def _jittered_rate_levels(n: int, *, seed: int = 13) -> list[float]:
+    """Уровни ставки с подъёмом + АПЕРИОДИЧНЫМ джиттером (низкая автокорреляция Δ).
+
+    Периодический (sin) джиттер даёт знак-переворачивающую автокорреляцию Δ —
+    инжектированный лаг тогда конкурирует с ложными лагами. LCG-джиттер делает
+    лаги слабо коррелированными, так что best_lag находит ИСТИННЫЙ лаг устойчиво.
+    """
+    lvl = 10.0
+    state = seed
+    out: list[float] = []
+    for _ in range(n):
+        state = (state * 1103515245 + 12345) % 2147483648
+        lvl += 0.3 + (state / 2147483648.0 - 0.5) * 0.4
+        out.append(lvl)
+    return out
+
+
+class TestComputeRateSensitivityHappy:
+    def _rate_levels(self, n: int) -> list[float]:
+        return _jittered_rate_levels(n)
+
+    def test_phrase_fills_x_y_z_and_sign(self) -> None:
+        n = 36
+        months = _months(n)
+        rate_levels = self._rate_levels(n)
+        lag = 3
+        beta = -0.05
+        seg_units = _synth_sales_units(rate_levels, lag=lag, beta=beta)
+        # Самый чувствительный bucket — "2-к 45-60" (сильнее реакция: beta·1.5).
+        bucket_units = {
+            ROOM_AREA_BUCKET_2K: _synth_sales_units(rate_levels, lag=lag, beta=beta * 1.5),
+        }
+        dispatcher = _Dispatcher(
+            months=months,
+            source_a_units=seg_units,
+            bucket_units=bucket_units,
+        )
+        macro = _macro_series(list(rate_levels), months)
+        with patch(_MACRO, return_value=macro), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec(district="Академический"))
+        assert isinstance(out, RateSensitivity)
+        assert out.confidence in ("high", "medium")
+        # Y: инжектированный лаг.
+        assert out.y_lag_months == lag
+        # X: отрицательный (продажи падают).
+        assert out.x_pct is not None and out.x_pct < 0
+        assert out.beta is not None and out.beta < 0
+        # Z: самый чувствительный bucket "2-к 45-60" → floor 45.
+        assert out.most_sensitive_bucket == ROOM_AREA_BUCKET_2K
+        assert out.z_area_floor == _BUCKET_AREA_FLOOR_M2[ROOM_AREA_BUCKET_2K]
+        # Фраза — полная форма, с заполненными X/Y/Z.
+        assert "снижались в среднем на" in out.phrase
+        assert f"через {lag} месяцев" in out.phrase
+        assert ROOM_AREA_BUCKET_2K in out.phrase
+        assert "более 45.0 м²" in out.phrase
+        # X в фразе — положительная магнитуда (= |x_pct|).
+        assert str(round(abs(out.x_pct), 1)) in out.phrase
+
+    def test_x_pct_matches_beta_exp_formula(self) -> None:
+        n = 30
+        months = _months(n)
+        rate_levels = self._rate_levels(n)
+        seg_units = _synth_sales_units(rate_levels, lag=1, beta=-0.04)
+        dispatcher = _Dispatcher(months=months, source_a_units=seg_units)
+        macro = _macro_series(list(rate_levels), months)
+        with patch(_MACRO, return_value=macro), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec())
+        assert out.beta is not None and out.x_pct is not None
+        expected = 100.0 * (math.exp(out.beta) - 1.0)
+        assert math.isclose(out.x_pct, expected, rel_tol=1e-9)
+
+
+class TestComputeRateSensitivityDegrade:
+    def test_thin_segment_insufficient(self) -> None:
+        # Слишком короткий ряд: < _MIN_OBS валидных Δln-точек → нет gate-лага.
+        n = 5
+        months = _months(n)
+        rate_levels = [10.0 + i for i in range(n)]
+        seg_units = _synth_sales_units(rate_levels, lag=1, beta=-0.05)
+        dispatcher = _Dispatcher(months=months, source_a_units=seg_units)
+        macro = _macro_series(list(map(float, rate_levels)), months)
+        with patch(_MACRO, return_value=macro), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec())
+        assert out.confidence == "low"
+        assert out.phrase == _PHRASE_INSUFFICIENT
+        assert out.x_pct is None and out.y_lag_months is None
+
+    def test_wrong_sign_degrades_to_insufficient(self) -> None:
+        # Продажи РАСТУТ при ↑ставки (β>0) — выживший неправильный знак → insufficient.
+        # Апериодический подъём ставки (LCG-джиттер) → позитивная связь держит знак
+        # на всех лагах (без sin-переворота), так что gate(slope<0) валит все лаги.
+        n = 30
+        months = _months(n)
+        lvl = 10.0
+        state = 11
+        rate_levels: list[float] = []
+        for _ in range(n):
+            state = (state * 1103515245 + 12345) % 2147483648
+            lvl += 0.2 + (state / 2147483648.0) * 0.3  # всегда положительный шаг
+            rate_levels.append(lvl)
+        seg_units = _synth_sales_units(rate_levels, lag=1, beta=+0.05)  # положительный
+        dispatcher = _Dispatcher(months=months, source_a_units=seg_units)
+        macro = _macro_series(list(rate_levels), months)
+        with patch(_MACRO, return_value=macro), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec())
+        assert out.confidence == "low"
+        assert out.phrase == _PHRASE_INSUFFICIENT
+
+    def test_ekb_wide_form_when_no_sensitive_bucket(self) -> None:
+        # Сегмент даёт валидный негативный β, но НИ один Source-B bucket не проходит
+        # gate (все плоские) → фраза деградирует к EKB-широкой форме (без 2-го предл.).
+        n = 36
+        months = _months(n)
+        rate_levels = _jittered_rate_levels(n)
+        seg_units = _synth_sales_units(rate_levels, lag=2, beta=-0.05)
+        flat = [500] * n  # все bucket-ряды плоские → нет реакции → нет gate-лага
+        dispatcher = _Dispatcher(
+            months=months,
+            source_a_units=seg_units,
+            default_units=flat,
+        )
+        macro = _macro_series(list(rate_levels), months)
+        with patch(_MACRO, return_value=macro), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec())
+        assert out.y_lag_months == 2
+        assert out.x_pct is not None and out.x_pct < 0
+        assert out.most_sensitive_bucket is None
+        assert out.z_area_floor is None
+        # EKB-широкая форма: 1-е предложение есть, 2-е («наиболее чувствительны») нет.
+        assert "снижались в среднем на" in out.phrase
+        assert "Наиболее чувствительны" not in out.phrase
+
+    def test_confounded_flag_flows_and_caps_confidence(self) -> None:
+        # Окно ряда пересекает шок 2022-02 → confounded=True; даже сильный фит → не 'high'.
+        n = 36
+        months = _months(n, end=dt.date(2022, 12, 1))  # окно охватывает 2022-02
+        assert any(m == dt.date(2022, 2, 1) for m in months)
+        rate_levels = _jittered_rate_levels(n)
+        seg_units = _synth_sales_units(rate_levels, lag=2, beta=-0.05)
+        bucket_units = {
+            ROOM_AREA_BUCKET_2K: _synth_sales_units(rate_levels, lag=2, beta=-0.07),
+        }
+        dispatcher = _Dispatcher(months=months, source_a_units=seg_units, bucket_units=bucket_units)
+        macro = _macro_series(list(rate_levels), months)
+        with patch(_MACRO, return_value=macro), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec())
+        assert out.confounded is True
+        assert out.confidence != "high"  # confounded окно срезает high
+
+    def test_graceful_empty_is_low(self) -> None:
+        # Пустые ряды (нет данных) → low, insufficient, не crash.
+        months: list[dt.date] = []
+        dispatcher = _Dispatcher(months=months, source_a_units=[])
+        with patch(_MACRO, return_value=[]), patch(_BUILD, side_effect=dispatcher):
+            out = compute_rate_sensitivity(MagicMock(), spec=SegmentSpec())
+        assert out.confidence == "low"
+        assert out.phrase == _PHRASE_INSUFFICIENT
+        assert out.n_obs == 0
+
+
+class TestRateSensitivityAsDict:
+    def test_serialises_and_rounds(self) -> None:
+        rs = RateSensitivity(
+            segment={"district": "X", "obj_class": None, "room_bucket": None, "price_bucket": None},
+            x_pct=-12.3456,
+            y_lag_months=3,
+            z_area_floor=45.0,
+            most_sensitive_bucket=ROOM_AREA_BUCKET_2K,
+            beta=-0.131234,
+            r2=0.456789,
+            n_obs=28,
+            shrinkage_weight=0.7361,
+            confounded=False,
+            confidence="high",
+            phrase="…",
+        )
+        d = rs.as_dict()
+        assert d["x_pct"] == -12.3
+        assert d["beta"] == -0.1312
+        assert d["r2"] == 0.4568
+        assert d["shrinkage_weight"] == 0.736
+        assert d["y_lag_months"] == 3
+        assert d["confidence"] == "high"
+
+    def test_none_numerics_survive(self) -> None:
+        rs = RateSensitivity(
+            segment={},
+            x_pct=None,
+            y_lag_months=None,
+            z_area_floor=None,
+            most_sensitive_bucket=None,
+            beta=None,
+            r2=None,
+            n_obs=0,
+            shrinkage_weight=0.0,
+            confounded=False,
+            confidence="low",
+            phrase=_PHRASE_INSUFFICIENT,
+        )
+        d = rs.as_dict()
+        assert d["x_pct"] is None
+        assert d["beta"] is None
+        assert d["phrase"] == _PHRASE_INSUFFICIENT