diff --git a/backend/app/services/forecasting/__init__.py b/backend/app/services/forecasting/__init__.py index c3c48e00..2be86bed 100644 --- a/backend/app/services/forecasting/__init__.py +++ b/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", ] diff --git a/backend/app/services/forecasting/rate_sensitivity.py b/backend/app/services/forecasting/rate_sensitivity.py new file mode 100644 index 00000000..6d77b54d --- /dev/null +++ b/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) diff --git a/backend/tests/services/forecasting/test_rate_sensitivity.py b/backend/tests/services/forecasting/test_rate_sensitivity.py new file mode 100644 index 00000000..677b3d14 --- /dev/null +++ b/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