feat(forecasting): wire Almon-ADL rate estimator into §9.6 consumers (#978)

Backtest (OOS directional hit-rate): single-best-lag compute_rate_sensitivity
is directionally noise (0.148 Source B EKB-wide, lag-unstable); the Almon
distributed-lag estimator (compute_district_rate_regression) is strictly less
noisy on every tier (0.407 Source B / 0.60 survivorship-free Source A,
lag-stable). Add a thin adapter compute_rate_regime_sensitivity mapping
DistributedLagFit onto the existing RateSensitivity contract (beta=long-run
sum-beta, confidence regression->medium / fallback->low, district=None->low and
no call) and repoint the three consumers (demand_normalization, product_scoring,
demand_supply_forecast). Magnitude bounded by the existing [0.5,1.2] clamp.
Reversible; compute_rate_sensitivity kept for the backtest. Consumer tests
repointed to the real Almon path (mutation-verified genuine) + adapter unit
tests + end-to-end fallback degradation. Forecasting suite 840 passed; ruff clean.

backend/app/services/forecasting/demand_normalization.py

@@ -57,7 +57,7 @@ from typing import Any, Literal
 from sqlalchemy.orm import Session
 
 from app.services.forecasting.macro_series import MonthlyMacro, get_monthly_macro
-from app.services.forecasting.rate_sensitivity import compute_rate_sensitivity
+from app.services.forecasting.regression import compute_rate_regime_sensitivity
 from app.services.forecasting.sales_series import SegmentSpec
 
 logger = logging.getLogger(__name__)
@@ -265,8 +265,8 @@ def compute_demand_normalization(
     """
     segment = spec.as_dict()
 
-    # ── 1. β §9.6 ──────────────────────────────────────────────────────────────
+    # ── 1. β §9.6 (Almon-ADL long-run multiplier via the validated #978 estimator)
-    sensitivity = compute_rate_sensitivity(db, spec=spec, months_back=months_back)
+    sensitivity = compute_rate_regime_sensitivity(db, spec=spec, months_back=months_back)
     beta = sensitivity.beta
 
     # ── 2. Средняя ставка окна наблюдения ──────────────────────────────────────

backend/app/services/forecasting/demand_supply_forecast.py

@@ -63,7 +63,7 @@ from app.schemas.parcel import CompetitorsRequest
 from app.services.forecasting.demand_normalization import compute_demand_normalization
 from app.services.forecasting.macro_coefficient import compute_macro_coefficient
 from app.services.forecasting.macro_series import MonthlyMacro, get_monthly_macro
-from app.services.forecasting.rate_sensitivity import compute_rate_sensitivity
+from app.services.forecasting.regression import compute_rate_regime_sensitivity
 from app.services.forecasting.sales_series import SegmentSpec
 from app.services.site_finder.competitors import get_competitors
 from app.services.site_finder.future_supply import compute_future_supply_pressure
@@ -499,7 +499,7 @@ def compute_demand_supply_forecast(
     macro_coef = compute_macro_coefficient(db, segment_profile=profile)
 
     # ── Один раз: §9.6 чувствительность — ТОЛЬКО для explain-фразы (НЕ арифметика)
-    sensitivity = compute_rate_sensitivity(db, spec=spec)
+    sensitivity = compute_rate_regime_sensitivity(db, spec=spec)
 
     out: list[DemandSupplyForecast] = []
     for h in horizon_list:

backend/app/services/forecasting/product_scoring.py

@@ -58,8 +58,8 @@ from sqlalchemy.orm import Session
 from app.schemas.parcel import CompetitorsRequest
 from app.services.forecasting.affordability import compute_affordability
 from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast
-from app.services.forecasting.rate_sensitivity import compute_rate_sensitivity
 from app.services.forecasting.recommendation import build_forecast_overlay
+from app.services.forecasting.regression import compute_rate_regime_sensitivity
 from app.services.forecasting.sales_series import SegmentSpec
 from app.services.site_finder.competitors import get_competitors
 from app.services.site_finder.future_supply import compute_future_supply_pressure
@@ -672,8 +672,10 @@ def compute_score_card(
         "infra_fit", _K_INFRA_FIT, "poi_score", _score_infra_fit(poi_sum, "medium")
     )
 
-    # ── mortgage_sensitivity ← §9.6 x_pct ──────────────────────────────────────
+    # ── mortgage_sensitivity ← §9.6 x_pct (Almon-ADL estimator, #978) ──────────
-    sensitivity = _safe_call("rate_sensitivity", lambda: compute_rate_sensitivity(db, spec=spec))
+    sensitivity = _safe_call(
+        "rate_sensitivity", lambda: compute_rate_regime_sensitivity(db, spec=spec)
+    )
     x_pct = sensitivity.x_pct if sensitivity is not None else None
     sens_conf: Confidence = sensitivity.confidence if sensitivity is not None else "low"
     scores[_K_MORTGAGE_SENSITIVITY] = _build(

backend/app/services/forecasting/regression.py

@@ -54,7 +54,7 @@ import numpy as np
 from sqlalchemy.orm import Session
 
 from app.services.forecasting.macro_series import get_monthly_macro
-from app.services.forecasting.rate_sensitivity import _delta
+from app.services.forecasting.rate_sensitivity import Confidence, RateSensitivity, _delta
 from app.services.forecasting.sales_series import (
     SegmentSpec,
     build_sales_series,
@@ -627,3 +627,150 @@ def compute_district_rate_regression(
         max_lag=max_lag,
         degree=degree,
     )
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# §9.6 production adapter — wraps the validated Almon-ADL estimator behind the
+# RateSensitivity contract the three §9.6 consumers already speak.
+# ──────────────────────────────────────────────────────────────────────────────
+#
+# WHY this exists: the out-of-sample backtest (backend/scripts/backtest_rate_
+# sensitivity.py) found the single-best-lag OLS (`rate_sensitivity.compute_rate_
+# sensitivity`) directionally NOISE OOS (hit-rate 0.148 on Source B EKB-wide,
+# worse than coin-flip), while this Almon distributed-lag estimator is strictly
+# less noisy on every tier (0.407 Source B / 0.60 survivorship-free Source A).
+# So the rate-regime discount (§9.4 demand_normalization), the mortgage_sensitivity
+# score (§14.2 product_scoring) and the explain-phrase (§9.8 demand_supply_forecast)
+# are repointed onto this estimator. REVERSIBLE: this adapter is the single seam —
+# reverting the 3 call-site imports restores the old path. compute_rate_sensitivity
+# stays as-is (still used by the backtest + its own tests).
+
+
+def _insufficient_sensitivity(segment: dict[str, str | None]) -> RateSensitivity:
+    """Low-confidence RateSensitivity with no usable β/x_pct (graceful degrade).
+
+    Used when there is no district to fit the district×class regression on
+    (spec.district is None): demand_normalization then degrades to a neutral
+    norm=1.0 (applied=False) and product_scoring's mortgage_sensitivity takes the
+    low-confidence path — honest, not invented. 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=0,
+        shrinkage_weight=0.0,
+        confounded=False,
+        confidence="low",
+        phrase=_PHRASE_INSUFFICIENT,
+    )
+
+
+def _fit_to_sensitivity(
+    fit: DistributedLagFit, *, segment: dict[str, str | None]
+) -> RateSensitivity:
+    """Map a DistributedLagFit (Almon-ADL) onto the §9.6 RateSensitivity contract.
+
+    Field mapping (see compute_rate_regime_sensitivity docstring for the rationale):
+      • beta          ← fit.coef        (LONG-RUN Σβ — see beta-semantics note below)
+      • x_pct         ← fit.x_pct
+      • y_lag_months  ← fit.best_lag_months
+      • phrase        ← fit.phrase
+      • r2 / n_obs    ← fit.r2 / fit.n
+      • confidence    ← 'regression' → "medium" (gated-OK but advisory-grade) |
+                        'fallback'   → "low"
+    Source-B-only outputs (z_area_floor, most_sensitive_bucket, confounded,
+    shrinkage_weight) have no analogue in a district×class distributed-lag fit
+    (no room×area bucketing here) → None / sensible defaults. PURE.
+
+    BETA SEMANTICS (important): `beta` here carries the Almon LONG-RUN multiplier
+    Σ_j β_j on Δln — the cumulative %-effect of a SUSTAINED +1pp regime shift, NOT
+    a single-lag slope. That is exactly the quantity demand_normalization wants for
+    a future-regime discount (exp(β·Δrate) over a sustained Δrate), and it stays
+    clamped to [0.5, 1.2] downstream so a large coef saturates rather than blows up.
+    """
+    confidence: Confidence = "medium" if fit.source == "regression" else "low"
+    return RateSensitivity(
+        segment=segment,
+        x_pct=fit.x_pct,
+        y_lag_months=fit.best_lag_months,
+        z_area_floor=None,
+        most_sensitive_bucket=None,
+        beta=fit.coef,
+        r2=fit.r2,
+        n_obs=fit.n,
+        shrinkage_weight=0.0,
+        confounded=False,
+        confidence=confidence,
+        phrase=fit.phrase,
+    )
+
+
+def compute_rate_regime_sensitivity(
+    db: Session,
+    *,
+    spec: SegmentSpec,
+    months_back: int = _DEFAULT_MONTHS_BACK,
+) -> RateSensitivity:
+    """§9.6 rate sensitivity for a market segment via the Almon-ADL estimator.
+
+    Thin adapter over `compute_district_rate_regression` (the validated #978
+    distributed-lag model) that returns the existing `RateSensitivity` dataclass so
+    the three §9.6 consumers (demand_normalization / product_scoring /
+    demand_supply_forecast) use it with NO body changes beyond the call site.
+    Replaces the OOS-noisy single-best-lag `compute_rate_sensitivity` in production.
+
+    Confidence is capped at "medium" even on a gate-passing fit: the §9.6 stack is
+    advisory until the engine is fully validated, so we never advertise "high".
+
+    Graceful degradation (NEVER crashes):
+      • spec.district is None → no district to fit the district×class regression on
+        → low-confidence result with beta=None / x_pct=None and the insufficient-
+        data phrase (demand_normalization → neutral, product_scoring → low). We do
+        NOT call compute_district_rate_regression with district=None (it requires a
+        str).
+      • compute_district_rate_regression is already graceful (returns a 'fallback'
+        DistributedLagFit on thin/failed data), but we still wrap it defensively and
+        degrade to the insufficient result on any unexpected error.
+
+    Args:
+        db: SQLAlchemy sync Session.
+        spec: target segment; `district` (and optionally `obj_class`) drive the fit.
+        months_back: series depth (defaults to _DEFAULT_MONTHS_BACK).
+
+    Returns:
+        RateSensitivity (always; phrase populated even when data is insufficient).
+    """
+    segment = spec.as_dict()
+
+    if spec.district is None:
+        logger.info(
+            "rate_regime_sensitivity: spec.district is None (segment=%s) → "
+            "insufficient (district×class regression needs a district)",
+            segment,
+        )
+        return _insufficient_sensitivity(segment)
+
+    try:
+        fit = compute_district_rate_regression(
+            db,
+            district=spec.district,
+            obj_class=spec.obj_class,
+            months_back=months_back,
+        )
+    except Exception:
+        # compute_district_rate_regression is graceful by contract; this guard only
+        # catches truly unexpected failures so the §9.6 consumers never crash on the
+        # rate channel. Log with traceback (never swallow silently), then degrade.
+        logger.exception(
+            "rate_regime_sensitivity: district regression raised (segment=%s) → "
+            "degrading to insufficient",
+            segment,
+        )
+        return _insufficient_sensitivity(segment)
+
+    return _fit_to_sensitivity(fit, segment=segment)

backend/tests/services/forecasting/test_demand_normalization.py

@@ -5,11 +5,18 @@
     режимы совпали (future==window) → 1.0; β None / β=0 → 1.0; клэмп на MIN при
     огромном разрыве; β>0-край → аплифт, но клэмп на MAX; rate_window_avg None → 1.0.
   • _window_avg_rate — среднее НЕпустых key_rate; все None → None.
-  • compute_demand_normalization (мок compute_rate_sensitivity + get_monthly_macro):
+  • compute_demand_normalization (мок compute_district_rate_regression +
-    надёжный β + более жёсткое будущее → coef<1 + applied=True; low-conf β → 1.0 +
+    get_monthly_macro): надёжный β + более жёсткое будущее → coef<1 + applied=True;
-    applied=False; недоступный β (None) → 1.0/False; пустой макро-ряд → 1.0/low;
+    fallback-фит (β=None/low) → 1.0 + applied=False; пустой макро-ряд → 1.0/low;
     confidence наследуется (не выше §9.6); future<window → аплифт >1; знак Δ.
 
+§9.6-источник β — теперь Almon-ADL `compute_district_rate_regression` (#978),
+обёрнутый адаптером `compute_rate_regime_sensitivity`. Тесты оркестратора патчат
+РЕАЛЬНЫЙ Almon-вход настоящим DistributedLagFit и прогоняют адаптер целиком (а не
+мокают уже-неиспользуемый single-lag compute_rate_sensitivity) → проверяем НОВЫЙ
+путь: 'regression'-фит → confidence='medium', β=coef; 'fallback'-фит → β=None,
+confidence='low' → нейтраль.
+
 ЗНАКОВАЯ ЛОГИКА (тестируем явно): β<0, future>window → β·(+)<0 → exp<1 → ДИСКОНТ
 (суть §9.4 — не тащить бумный темп в более жёсткий режим). ЧЕСТНОСТЬ: applied=False,
 когда β ненадёжен/недоступен (нейтраль 1.0 без выдуманного дисконта).
@@ -34,10 +41,14 @@ from app.services.forecasting.demand_normalization import (
     normalization_factor,
 )
 from app.services.forecasting.macro_series import MonthlyMacro
-from app.services.forecasting.rate_sensitivity import RateSensitivity
+from app.services.forecasting.regression import DistributedLagFit
 from app.services.forecasting.sales_series import SegmentSpec
 
-_SENS = "app.services.forecasting.demand_normalization.compute_rate_sensitivity"
+# Адаптер §9.6 (compute_rate_regime_sensitivity) импортирован в namespace
+# demand_normalization; он зовёт compute_district_rate_regression в namespace
+# regression. Патчим РЕАЛЬНЫЙ Almon-вход и прогоняем адаптер целиком (а не мокаем
+# уже-неиспользуемый single-lag compute_rate_sensitivity — это был бы false-green).
+_REG = "app.services.forecasting.regression.compute_district_rate_regression"
 _MACRO = "app.services.forecasting.demand_normalization.get_monthly_macro"
 
 
@@ -73,25 +84,31 @@ def _macro(months: list[dt.date], rates: list[float | None]) -> list[MonthlyMacr
     return out
 
 
-def _sensitivity(
+def _reg_fit(
     *,
-    beta: float | None,
+    coef: float | None,
-    confidence: str,
+    source: str = "regression",
     segment: dict[str, str | None] | None = None,
-) -> RateSensitivity:
+) -> DistributedLagFit:
-    """RateSensitivity-заглушка §9.6 с нужным β и confidence (прочее не важно §9.4)."""
+    """РЕАЛЬНЫЙ Almon DistributedLagFit-вход §9.6 (#978), который адаптер мапит в β.
-    return RateSensitivity(
+
-        segment=segment or {},
+    'regression'-source → адаптер выдаёт confidence='medium', beta=coef; 'fallback'
-        x_pct=None if beta is None else 100.0 * (math.exp(beta) - 1.0),
+    → beta=None/confidence='low' (coef/x_pct=None — контракт build_fit_result). Это
-        y_lag_months=None if beta is None else 3,
+    НАСТОЯЩИЙ контракт DistributedLagFit→адаптер, не hand-faked attribute bag.
-        z_area_floor=None,
+    """
-        most_sensitive_bucket=None,
+    gated = source == "regression"
-        beta=beta,
+    return DistributedLagFit(
-        r2=None if beta is None else 0.5,
+        segment=segment or {"district": "X", "obj_class": None},
-        n_obs=0 if beta is None else 28,
+        best_lag_months=3 if gated else None,
-        shrinkage_weight=0.0 if beta is None else 0.7,
+        coef=coef if gated else None,
-        confounded=False,
+        x_pct=None if (coef is None or not gated) else 100.0 * (math.exp(coef) - 1.0),
-        confidence=confidence,  # type: ignore[arg-type]
+        r2=0.5 if gated else 0.05,
+        n=40 if gated else 12,
+        per_lag_coef=None,
+        hac_se=None,
+        hac_bandwidth=None,
+        almon_degree=2,
+        source=source,
         phrase="…",
     )
 
@@ -181,47 +198,52 @@ class TestWindowAvgRate:
 
 
 class TestComputeDemandNormalizationApplied:
-    def test_high_conf_beta_higher_future_discounts_and_applies(self) -> None:
+    def test_gated_beta_higher_future_discounts_and_applies(self) -> None:
-        # Надёжный β<0 + future(18) > window(avg≈8) → coef<1, applied=True.
+        # 'regression'-фит (β=coef<0) + future(18) > window(avg≈8) → coef<1, applied=True.
         n = 12
         months = _months(n)
         macro = _macro(months, [8.0] * n)  # окно «бума» — низкая ставка 8
-        sens = _sensitivity(beta=-0.03, confidence="high")
+        fit = _reg_fit(coef=-0.03, source="regression")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
             out = compute_demand_normalization(
                 MagicMock(), spec=SegmentSpec(district="Академический"), rate_future=18.0
             )
         assert isinstance(out, DemandNormalization)
         assert out.applied is True
         assert out.coefficient < _NORM_NEUTRAL  # дисконт: бумный темп срезан
-        assert out.beta == -0.03
+        assert out.beta == -0.03  # β = long-run Σβ из Almon-фита (= coef)
         assert out.rate_window_avg == 8.0
         assert out.rate_delta == 18.0 - 8.0
-        assert out.confidence == "high"
+        # Almon-адаптер кэпит confidence на 'medium' (advisory-grade, никогда 'high').
+        assert out.confidence == "medium"
         # Сверяем с pure-формулой (clamp(exp(β·Δ))).
         assert out.coefficient == normalization_factor(-0.03, 18.0, 8.0)
 
-    def test_medium_conf_lower_future_uplifts(self) -> None:
+    def test_gated_beta_lower_future_uplifts(self) -> None:
         # Наблюдали при жёсткой ставке (window≈16), будущее мягче (8) → аплифт >1.
         n = 12
         months = _months(n)
         macro = _macro(months, [16.0] * n)
-        sens = _sensitivity(beta=-0.02, confidence="medium")
+        fit = _reg_fit(coef=-0.02, source="regression")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=8.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=8.0
+            )
         assert out.applied is True
         assert out.coefficient > _NORM_NEUTRAL
         assert out.rate_delta == 8.0 - 16.0
-        assert out.confidence == "medium"  # наследуется от §9.6
+        assert out.confidence == "medium"  # наследуется от §9.6 (capped)
 
     def test_confidence_capped_at_sensitivity(self) -> None:
-        # Coef не «надёжнее» своего β: confidence ровно = §9.6 confidence.
+        # Coef не «надёжнее» своего β: confidence ровно = §9.6 confidence (medium).
         n = 12
         months = _months(n)
         macro = _macro(months, [10.0] * n)
-        sens = _sensitivity(beta=-0.04, confidence="medium")
+        fit = _reg_fit(coef=-0.04, source="regression")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=15.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=15.0
+            )
         assert out.confidence == "medium"
         assert out.applied is True
 
@@ -231,38 +253,45 @@ class TestComputeDemandNormalizationApplied:
         n = 12
         months = _months(n)
         macro = _macro(months, [12.0] * n)
-        sens = _sensitivity(beta=-0.05, confidence="high")
+        fit = _reg_fit(coef=-0.05, source="regression")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=12.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=12.0
+            )
         assert out.coefficient == _NORM_NEUTRAL
         assert out.applied is True  # коррекция оценена (Δ≈0), не деградация
 
 
 class TestComputeDemandNormalizationDegrade:
-    def test_low_conf_beta_neutral_not_applied(self) -> None:
+    def test_fallback_fit_neutral_not_applied(self) -> None:
-        # §9.6 confidence='low' (β ненадёжен) → нейтраль 1.0, applied=False, low.
+        # 'fallback'-фит §9.6 (gate провален → β=None, confidence='low') → нейтраль
-        # Честность: НЕ переносим бумный темп, но и НЕ выдумываем дисконт.
+        # 1.0, applied=False, low. Честность: НЕ переносим бумный темп, но и НЕ
+        # выдумываем дисконт. Это ОСНОВНОЙ контракт fallback→neutral (end-to-end).
         n = 12
         months = _months(n)
         macro = _macro(months, [8.0] * n)
-        sens = _sensitivity(beta=-0.03, confidence="low")
+        fit = _reg_fit(coef=None, source="fallback")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=18.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=18.0
+            )
         assert out.coefficient == _NORM_NEUTRAL
         assert out.applied is False
         assert out.confidence == "low"
         # rate_delta всё равно заполнен для explainability (оба конца известны).
         assert out.rate_delta == 18.0 - 8.0
-        assert out.beta == -0.03  # β сохранён (виден), но не применён
+        assert out.beta is None  # fallback → adapter выдаёт beta=None
 
-    def test_beta_none_neutral_not_applied(self) -> None:
+    def test_district_none_neutral_not_applied(self) -> None:
-        # β недоступен (§9.6 не дал валидного лага) → нейтраль 1.0, applied=False.
+        # spec.district=None → адаптер не зовёт регрессию (нет района) → β=None,
+        # low → нейтраль 1.0, applied=False. Patching get_monthly_macro достаточно;
+        # compute_district_rate_regression НЕ должен вызываться вовсе.
         n = 12
         months = _months(n)
         macro = _macro(months, [9.0] * n)
-        sens = _sensitivity(beta=None, confidence="low")
+        with patch(_REG) as reg_mock, patch(_MACRO, return_value=macro):
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
             out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=20.0)
+        reg_mock.assert_not_called()  # district=None → регрессию не зовём
         assert out.coefficient == _NORM_NEUTRAL
         assert out.applied is False
         assert out.beta is None
@@ -270,9 +299,11 @@ class TestComputeDemandNormalizationDegrade:
 
     def test_graceful_empty_macro_is_neutral_low(self) -> None:
         # Пустой макро-ряд → rate_window_avg=None → нейтраль 1.0, applied=False, low.
-        sens = _sensitivity(beta=-0.05, confidence="high")  # даже надёжный β
+        fit = _reg_fit(coef=-0.05, source="regression")  # даже надёжный β
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=[]):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=[]):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=15.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=15.0
+            )
         assert out.coefficient == _NORM_NEUTRAL
         assert out.applied is False
         assert out.confidence == "low"
@@ -284,9 +315,11 @@ class TestComputeDemandNormalizationDegrade:
         n = 12
         months = _months(n)
         macro = _macro(months, [None] * n)
-        sens = _sensitivity(beta=-0.05, confidence="high")
+        fit = _reg_fit(coef=-0.05, source="regression")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=15.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=15.0
+            )
         assert out.coefficient == _NORM_NEUTRAL
         assert out.applied is False
         assert out.rate_window_avg is None
@@ -296,9 +329,11 @@ class TestComputeDemandNormalizationDegrade:
         n = 12
         months = _months(n)
         macro = _macro(months, [5.0] * n)
-        sens = _sensitivity(beta=-0.4, confidence="high")
+        fit = _reg_fit(coef=-0.4, source="regression")
-        with patch(_SENS, return_value=sens), patch(_MACRO, return_value=macro):
+        with patch(_REG, return_value=fit), patch(_MACRO, return_value=macro):
-            out = compute_demand_normalization(MagicMock(), spec=SegmentSpec(), rate_future=30.0)
+            out = compute_demand_normalization(
+                MagicMock(), spec=SegmentSpec(district="X"), rate_future=30.0
+            )
         assert _NORM_MIN <= out.coefficient <= _NORM_MAX
         assert out.coefficient == _NORM_MIN  # огромный разрыв → пол
 

backend/tests/services/forecasting/test_demand_supply_forecast.py

@@ -50,6 +50,7 @@ from app.services.forecasting.demand_supply_forecast import (
     hold_last_rate,
 )
 from app.services.forecasting.macro_series import MonthlyMacro
+from app.services.forecasting.regression import DistributedLagFit
 from app.services.forecasting.sales_series import SegmentSpec
 
 # Пути патча reused-сервисов (импортированы в модуль demand_supply_forecast).
@@ -57,7 +58,12 @@ _MACRO = "app.services.forecasting.demand_supply_forecast.get_monthly_macro"
 _METRICS = "app.services.forecasting.demand_supply_forecast.compute_market_metrics"
 _NORM = "app.services.forecasting.demand_supply_forecast.compute_demand_normalization"
 _MACRO_COEF = "app.services.forecasting.demand_supply_forecast.compute_macro_coefficient"
-_SENS = "app.services.forecasting.demand_supply_forecast.compute_rate_sensitivity"
+# §9.6 для explain-фразы идёт через адаптер compute_rate_regime_sensitivity (в
+# namespace demand_supply_forecast), который зовёт compute_district_rate_regression
+# (Almon-ADL #978) в namespace regression. Патчим РЕАЛЬНЫЙ Almon-вход настоящим
+# DistributedLagFit и прогоняем адаптер целиком → фраза проходит НОВЫЙ путь (а не
+# мокаем уже-неиспользуемый single-lag compute_rate_sensitivity — был бы false-green).
+_SENS = "app.services.forecasting.regression.compute_district_rate_regression"
 _SUPPLY = "app.services.forecasting.demand_supply_forecast.compute_future_supply_pressure"
 _COMPETITORS = "app.services.forecasting.demand_supply_forecast.get_competitors"
 
@@ -484,13 +490,27 @@ def _macro_coef_stub(*, coefficient: float = 1.1, confidence: str = "high") -> M
 
 def _sens_stub(
     *, beta: float = -0.5, x_pct: float = -40.0, phrase: str = "при росте ставки …"
-) -> MagicMock:
+) -> DistributedLagFit:
-    """Стаб §9.6: несёт DISTINCTIVE beta/x_pct — они НЕ должны влиять на спрос."""
+    """РЕАЛЬНЫЙ Almon DistributedLagFit-вход §9.6 (#978) для адаптера.
-    m = MagicMock()
+
-    m.beta = beta
+    Несёт DISTINCTIVE coef(=beta)/x_pct — адаптер прокинет их в RateSensitivity, но
-    m.x_pct = x_pct
+    demand_supply_forecast берёт из §9.6 ТОЛЬКО .phrase (β/x_pct НЕ должны влиять на
-    m.phrase = phrase
+    спрос — он идёт через §9.4). 'regression'-source → адаптер выдаёт phrase=fit.phrase.
-    return m
+    """
+    return DistributedLagFit(
+        segment={"district": "Академический", "obj_class": "комфорт"},
+        best_lag_months=3,
+        coef=beta,
+        x_pct=x_pct,
+        r2=0.5,
+        n=40,
+        per_lag_coef=None,
+        hac_se=None,
+        hac_bandwidth=None,
+        almon_degree=2,
+        source="regression",
+        phrase=phrase,
+    )
 
 
 def _supply_stub(

backend/tests/services/forecasting/test_product_scoring.py

@@ -55,6 +55,7 @@ from app.services.forecasting.product_scoring import (
     _weighted_overall,
     compute_score_card,
 )
+from app.services.forecasting.regression import DistributedLagFit
 from app.services.forecasting.sales_series import SegmentSpec
 
 # Пути патча (backing-сервисы импортированы в namespace product_scoring).
@@ -65,7 +66,12 @@ _P_FSP = f"{_MOD}.compute_future_supply_pressure"
 _P_COMPETITORS = f"{_MOD}.get_competitors"
 _P_AFFORD = f"{_MOD}.compute_affordability"
 _P_POI = f"{_MOD}.compute_poi_weighted_top7"
-_P_SENS = f"{_MOD}.compute_rate_sensitivity"
+# §9.6 теперь идёт через адаптер compute_rate_regime_sensitivity (в namespace
+# product_scoring), который зовёт compute_district_rate_regression (Almon-ADL #978)
+# в namespace regression. Патчим РЕАЛЬНЫЙ Almon-вход настоящим DistributedLagFit и
+# прогоняем адаптер целиком → mortgage_sensitivity тестит НОВЫЙ путь (а не мокает
+# уже-неиспользуемый single-lag compute_rate_sensitivity — это был бы false-green).
+_P_REG = "app.services.forecasting.regression.compute_district_rate_regression"
 _P_OVERLAY = f"{_MOD}.build_forecast_overlay"
 
 
@@ -661,11 +667,28 @@ def _poi_response_stub(weights: list[float]) -> MagicMock:
     return r
 
 
-def _sens_stub(x_pct: float | None = -5.0, confidence: str = "medium") -> MagicMock:
+def _sens_stub(x_pct: float | None = -5.0, source: str = "regression") -> DistributedLagFit:
-    s = MagicMock()
+    """РЕАЛЬНЫЙ Almon DistributedLagFit-вход §9.6 (#978) для адаптера → mortgage_sens.
-    s.x_pct = x_pct
+
-    s.confidence = confidence
+    'regression'-source → адаптер выдаёт confidence='medium' + x_pct=fit.x_pct;
-    return s
+    'fallback' → x_pct=None / confidence='low' (build_fit_result обнуляет coef/x_pct
+    в fallback). Это НАСТОЯЩИЙ контракт DistributedLagFit→адаптер, не MagicMock-bag.
+    """
+    gated = source == "regression"
+    return DistributedLagFit(
+        segment={"district": "X", "obj_class": None},
+        best_lag_months=3 if gated else None,
+        coef=-0.05 if gated else None,
+        x_pct=x_pct if gated else None,
+        r2=0.5 if gated else 0.05,
+        n=40 if gated else 12,
+        per_lag_coef=None,
+        hac_se=None,
+        hac_bandwidth=None,
+        almon_degree=2,
+        source=source,
+        phrase="…",
+    )
 
 
 def _overlay_stub(
@@ -718,7 +741,7 @@ def _patch_all(
         patch(_P_COMPETITORS, return_value=competitors_rv),
         patch(_P_AFFORD, return_value=afford if afford is not None else _afford_stub()),
         patch(_P_POI, return_value=poi if poi is not None else _poi_response_stub([0.03, 0.02])),
-        patch(_P_SENS, return_value=sens if sens is not None else _sens_stub()),
+        patch(_P_REG, return_value=sens if sens is not None else _sens_stub()),
         patch(_P_OVERLAY, return_value=overlay_rv),
     ]
     return _MultiPatch(patchers)
@@ -737,7 +760,9 @@ class _MultiPatch:
             p.stop()
 
 
-_SPEC = SegmentSpec(obj_class="комфорт", room_bucket="2-к 45-60")
+# district задан → §9.6 Almon-адаптер реально фитит район×класс (без района адаптер
+# короткозамыкает в low/None — тогда mortgage_sensitivity не упражнял бы регрессию).
+_SPEC = SegmentSpec(obj_class="комфорт", room_bucket="2-к 45-60", district="Академический")
 _CAD = "66:41:0303161:123"
 
 
@@ -880,7 +905,7 @@ class TestComputeScoreCardGraceful:
             patch(_P_COMPETITORS, side_effect=boom),
             patch(_P_AFFORD, side_effect=boom),
             patch(_P_POI, side_effect=boom),
-            patch(_P_SENS, side_effect=boom),
+            patch(_P_REG, side_effect=boom),
             patch(_P_OVERLAY, side_effect=boom),
         ):
             card = compute_score_card(db, spec=_SPEC, district="X", cad_num=_CAD)
@@ -908,3 +933,61 @@ class TestComputeScoreCardGraceful:
         assert card_missing.overall is not None and card_full.overall is not None
         # overall остаётся валидным средним доступных (в [0,1]), не схлопывается к 0.
         assert 0.0 <= card_missing.overall <= 1.0
+
+
+# ── §9.6 mortgage_sensitivity через Almon-ADL адаптер (#978, НОВЫЙ путь) ────────
+
+
+class TestMortgageSensitivityViaRegimeAdapter:
+    """mortgage_sensitivity берёт x_pct/confidence из compute_rate_regime_sensitivity
+    (адаптер над Almon-ADL #978). Проверяем НОВЫЙ контракт сквозь реальный адаптер:
+    'regression'-фит → x_pct surfaced + medium; 'fallback'-фит → low-conf путь."""
+
+    def test_regression_fit_surfaces_x_pct_and_medium(self) -> None:
+        # 'regression' DistributedLagFit (x_pct=-12.0) → mortgage_sensitivity backed
+        # (value не None, инвертирован по магнитуде) + confidence 'medium' (capped).
+        with _patch_all(sens=_sens_stub(x_pct=-12.0, source="regression")):
+            card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
+        ms = card.scores["mortgage_sensitivity"]
+        assert ms.value is not None
+        assert 0.0 <= ms.value <= 1.0
+        assert ms.confidence == "medium"  # Almon-адаптер кэпит ≤ medium
+
+    def test_more_negative_x_pct_lowers_score(self) -> None:
+        # Инверсия сохранена сквозь адаптер: чувствительнее (отрицательнее x_pct) →
+        # НИЖЕ скор (mortgage_sensitivity high-bad → low-score).
+        with _patch_all(sens=_sens_stub(x_pct=-2.0, source="regression")):
+            mild = compute_score_card(
+                _db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
+            ).scores["mortgage_sensitivity"]
+        with _patch_all(sens=_sens_stub(x_pct=-25.0, source="regression")):
+            severe = compute_score_card(
+                _db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD
+            ).scores["mortgage_sensitivity"]
+        assert mild.value is not None and severe.value is not None
+        assert severe.value < mild.value
+
+    def test_fallback_fit_uses_low_confidence_path(self) -> None:
+        # 'fallback' DistributedLagFit → адаптер выдаёт x_pct=None + confidence='low'
+        # → mortgage_sensitivity value None (unavailable), честная деградация.
+        with _patch_all(sens=_sens_stub(source="fallback")):
+            card = compute_score_card(_db_with_centroid(), spec=_SPEC, district="X", cad_num=_CAD)
+        ms = card.scores["mortgage_sensitivity"]
+        assert ms.value is None
+        assert ms.confidence == "low"
+        # Карта всё равно собрана, прочие скоры backed.
+        assert card.scores["demand"].value is not None
+
+    def test_district_none_spec_degrades_low(self) -> None:
+        # spec.district=None → адаптер короткозамыкает (регрессию НЕ зовёт) → x_pct
+        # None + low → mortgage_sensitivity unavailable. compute_district_rate_
+        # regression не должен вызываться ни разу.
+        no_district = SegmentSpec(obj_class="комфорт", room_bucket="2-к 45-60")
+        with _patch_all():
+            with patch(_P_REG) as reg_mock:
+                card = compute_score_card(
+                    _db_with_centroid(), spec=no_district, district="X", cad_num=_CAD
+                )
+        reg_mock.assert_not_called()
+        assert card.scores["mortgage_sensitivity"].value is None
+        assert card.scores["mortgage_sensitivity"].confidence == "low"

backend/tests/services/forecasting/test_regression.py

@@ -22,6 +22,7 @@ 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")
 
@@ -29,6 +30,8 @@ import numpy as np
 import pytest
 
 from app.services.forecasting import regression as reg
+from app.services.forecasting.rate_sensitivity import RateSensitivity
+from app.services.forecasting.sales_series import SegmentSpec
 
 # --------------------------------------------------------------------------- #
 # Synthetic-series helpers — inject a KNOWN distributed-lag effect
@@ -511,3 +514,173 @@ class TestComputeDistrictRateRegression:
         )
         assert res.source == "fallback"
         assert res.phrase == reg._PHRASE_INSUFFICIENT
+
+
+# --------------------------------------------------------------------------- #
+# compute_rate_regime_sensitivity — production adapter (DistributedLagFit →
+# RateSensitivity) wiring the validated Almon-ADL estimator into the §9.6 consumers
+# --------------------------------------------------------------------------- #
+
+_ADAPT_REG = "app.services.forecasting.regression.compute_district_rate_regression"
+
+
+def _fit(
+    *,
+    coef: float | None,
+    x_pct: float | None,
+    best_lag: int | None,
+    r2: float | None,
+    n: int,
+    source: str,
+    phrase: str = "phrase",
+    segment: dict[str, str | None] | None = None,
+) -> reg.DistributedLagFit:
+    """A REAL DistributedLagFit (the dataclass the adapter actually receives).
+
+    Not a hand-faked attribute bag: building the real frozen dataclass guarantees
+    the adapter mapping is tested against the true #978 contract and breaks loudly
+    if a field is renamed.
+    """
+    return reg.DistributedLagFit(
+        segment=segment or {"district": "Академический", "obj_class": None},
+        best_lag_months=best_lag,
+        coef=coef,
+        x_pct=x_pct,
+        r2=r2,
+        n=n,
+        per_lag_coef=None,
+        hac_se=None,
+        hac_bandwidth=None,
+        almon_degree=2,
+        source=source,
+        phrase=phrase,
+    )
+
+
+class TestComputeRateRegimeSensitivity:
+    def test_regression_source_maps_fields_and_medium(self) -> None:
+        # 'regression' (gate passed) → beta==coef (long-run Σβ), x_pct/y_lag/phrase
+        # mapped through, confidence 'medium' (advisory-grade, never 'high').
+        fit = _fit(
+            coef=-0.12,
+            x_pct=-11.3,
+            best_lag=2,
+            r2=0.42,
+            n=40,
+            source="regression",
+            phrase="спрос снижается …",
+        )
+        with patch(_ADAPT_REG, return_value=fit) as reg_mock:
+            out = reg.compute_rate_regime_sensitivity(
+                MagicMock(), spec=SegmentSpec(district="Академический", obj_class="комфорт")
+            )
+        assert isinstance(out, RateSensitivity)
+        assert out.beta == -0.12  # beta ← fit.coef (Almon long-run multiplier)
+        assert out.x_pct == -11.3
+        assert out.y_lag_months == 2
+        assert out.phrase == "спрос снижается …"
+        assert out.confidence == "medium"
+        assert out.r2 == 0.42
+        assert out.n_obs == 40
+        # Source-B-only fields have no analogue in a district×class fit.
+        assert out.z_area_floor is None
+        assert out.most_sensitive_bucket is None
+        # Adapter forwarded district + obj_class to the district regression.
+        call = reg_mock.call_args
+        assert call.kwargs["district"] == "Академический"
+        assert call.kwargs["obj_class"] == "комфорт"
+
+    def test_fallback_source_maps_none_and_low(self) -> None:
+        # 'fallback' (degrade) → beta is None, x_pct None, confidence 'low'; phrase
+        # is the insufficient phrase carried from the fit.
+        fit = _fit(
+            coef=None,
+            x_pct=None,
+            best_lag=None,
+            r2=0.04,
+            n=12,
+            source="fallback",
+            phrase=reg._PHRASE_INSUFFICIENT,
+        )
+        with patch(_ADAPT_REG, return_value=fit):
+            out = reg.compute_rate_regime_sensitivity(
+                MagicMock(), spec=SegmentSpec(district="Академический")
+            )
+        assert out.beta is None
+        assert out.x_pct is None
+        assert out.y_lag_months is None
+        assert out.confidence == "low"
+        assert out.phrase == reg._PHRASE_INSUFFICIENT
+        # Diagnostic numbers still surface (n carried through).
+        assert out.n_obs == 12
+
+    def test_district_none_short_circuits_low_no_call(self) -> None:
+        # spec.district is None → adapter must NOT call the district regression (it
+        # requires a str) and degrades to a low-confidence, beta=None result.
+        with patch(_ADAPT_REG) as reg_mock:
+            out = reg.compute_rate_regime_sensitivity(
+                MagicMock(), spec=SegmentSpec(obj_class="комфорт")
+            )
+        reg_mock.assert_not_called()
+        assert out.beta is None
+        assert out.x_pct is None
+        assert out.confidence == "low"
+        assert out.phrase == reg._PHRASE_INSUFFICIENT
+        # Segment dict still reflects the spec (shape preserved for consumers).
+        assert out.segment == SegmentSpec(obj_class="комфорт").as_dict()
+
+    def test_internal_failure_degrades_not_crash(self) -> None:
+        # Defensive guard: if the (graceful-by-contract) district regression still
+        # raises unexpectedly, the adapter degrades to insufficient, never crashes.
+        with patch(_ADAPT_REG, side_effect=RuntimeError("boom")):
+            out = reg.compute_rate_regime_sensitivity(
+                MagicMock(), spec=SegmentSpec(district="Академический")
+            )
+        assert out.beta is None
+        assert out.confidence == "low"
+        assert out.phrase == reg._PHRASE_INSUFFICIENT
+
+    def test_segment_carries_full_spec_shape(self) -> None:
+        # The returned segment mirrors spec.as_dict() (4-axis shape the consumers
+        # already serialise), not the 2-key regression segment.
+        fit = _fit(coef=-0.05, x_pct=-4.9, best_lag=1, r2=0.3, n=33, source="regression")
+        spec = SegmentSpec(obj_class="бизнес", room_bucket="2-к 45-60", district="X")
+        with patch(_ADAPT_REG, return_value=fit):
+            out = reg.compute_rate_regime_sensitivity(MagicMock(), spec=spec)
+        assert out.segment == spec.as_dict()
+        assert set(out.segment) == {"obj_class", "room_bucket", "district", "price_bucket"}
+
+    def test_uses_real_fit_via_synthetic_series_end_to_end(self, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+        # Strongest contract check: run the adapter over the REAL compute_district_
+        # rate_regression on a synthetic lag-2 series (no faking of the fit object) →
+        # confidence 'medium', beta == the real long-run coef, x_pct negative.
+        n = 60
+        months = _months(n)
+        xdelta = _aperiodic_rate_deltas(n, seed=13)
+        levels: list[float] = []
+        acc = 10.0
+        for d in xdelta:
+            acc += d
+            levels.append(acc)
+        macro = [_FakeMacro(months[i], levels[i]) for i in range(n)]
+        beta_scalar, lag = -0.05, 2
+        ln_u = math.log(1000.0)
+        units: list[int] = []
+        for t in range(n):
+            if t > 0:
+                src = xdelta[t - lag] if t - lag >= 0 else 0.0
+                ln_u += beta_scalar * src
+            units.append(max(1, round(math.exp(ln_u))))
+        sales = _FakeSales(months, units)
+        monkeypatch.setattr(reg, "get_monthly_macro", lambda db, months_back: macro)
+        monkeypatch.setattr(reg, "build_sales_series", lambda db, spec, source, months_back: sales)
+
+        out = reg.compute_rate_regime_sensitivity(
+            object(),  # type: ignore[arg-type]
+            spec=SegmentSpec(district="Академический"),
+            months_back=n,
+        )
+        assert out.confidence == "medium"
+        assert out.beta is not None and out.beta < 0
+        assert out.x_pct is not None and out.x_pct < 0
+        assert out.y_lag_months in (1, 2, 3)