diff --git a/backend/app/services/forecasting/__init__.py b/backend/app/services/forecasting/__init__.py index 2be86bed..2157b17f 100644 --- a/backend/app/services/forecasting/__init__.py +++ b/backend/app/services/forecasting/__init__.py @@ -9,7 +9,7 @@ • macro_series (#951b) — monthly макро-ряд + классификатор режима ставки (X-ось §9.6). • sales_series (#951c) — monthly ряд продаж по сегменту (Y-ось §9.6). • rate_sensitivity (#951d) — §9.6 чувствительность продаж к key_rate (CORE, ADVISORY). - • macro_coefficient (later PR) — §9.5 макро-коэффициент. + • macro_coefficient (#951e) — §9.5 макро-коэффициент (композитный множитель, ADVISORY). Источники данных: • макро — таблица macro_indicator через reader site_finder/macro.py (reuse). @@ -18,6 +18,17 @@ from __future__ import annotations +from app.services.forecasting.macro_coefficient import ( + MacroCoefficient, + assemble_coefficient, + compute_macro_coefficient, + f_issuance, + f_mortgage_rate, + f_overdue, + f_rate, + renormalize_contributions, + segment_steepness, +) from app.services.forecasting.macro_series import ( MonthlyMacro, classify_regime, @@ -43,14 +54,21 @@ from app.services.forecasting.sales_series import ( ) __all__ = [ + "MacroCoefficient", "MonthlyMacro", "RateSensitivity", "SalesSeries", "SegmentSpec", + "assemble_coefficient", "best_lag", "build_sales_series", "classify_regime", + "compute_macro_coefficient", "compute_rate_sensitivity", + "f_issuance", + "f_mortgage_rate", + "f_overdue", + "f_rate", "fill_month_grid", "get_monthly_macro", "is_confounded_window", @@ -58,6 +76,8 @@ __all__ = [ "macro_at_lag", "ols_slope_r2", "price_bucket_of", + "renormalize_contributions", "room_area_bucket_of", + "segment_steepness", "shrink", ] diff --git a/backend/app/services/forecasting/macro_coefficient.py b/backend/app/services/forecasting/macro_coefficient.py new file mode 100644 index 00000000..8b0068f0 --- /dev/null +++ b/backend/app/services/forecasting/macro_coefficient.py @@ -0,0 +1,778 @@ +"""§9.5 макроэкономический коэффициент — детерминированный композитный множитель. + +#951 (Site Finder v2 / «GG-форсайт» ТЗ §9.5), EPIC 7 «Чувствительность к ключевой +ставке», sub-PR 4 (#951e). §9.5 поправляет прогноз спроса на текущий МАКРО-РЕЖИМ: +композитный множитель, центрированный в 1.0 (1.0 = «макрорежим нейтрален к спросу»; +<1 = режим давит на спрос; >1 = режим поддерживает спрос), направленный по сегменту +рынка. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (арифметика на именованных константах). + +ЭВРИСТИКА, НЕ ФИТ: веса и шаги-крутизны — ИМЕНОВАННЫЕ КОНСТАНТЫ (экспертная оценка), +а НЕ обученная модель. Поэтому модуль уезжает в прод сейчас и уточняется позже на +out-of-sample (как и §9.6 rate_sensitivity — тот тоже advisory до бэктеста PR6). + +ADVISORY-СТАТУС: коэффициент СОВЕТУЮЩИЙ и НЕ подключён ни к одному эндпоинту в этом +PR. Цифры пригодны для explainability/прототипа, но НЕ для production-решений, пока +веса не валидированы. + +Формула (центр 1.0; каждый sub-factor — направленный нудж в [-1, 1] ДО веса): + + macro_coef = clamp( + 1.0 + + W_RATE · f_rate(тренд key_rate) # rate↑ → <0 + + W_MORTG_RATE · f_mortgage_rate(Δ mortgage_rate_weighted) # ↑ → <0 + + W_ISSUANCE · f_issuance(Δ count & Δ volume ИЖК) # выдачи↓ → <0 + + W_OVERDUE · f_overdue(overdue / debt) # доля↑ → <0 + + W_GOVSUPPORT · None # DEGRADED — нет поля в MonthlyMacro → не участвует + + W_INCOME · None # DEGRADED + + W_INFLATION · None # DEGRADED + + W_CONFIDENCE · None, # DEGRADED + _MACRO_COEF_MIN, _MACRO_COEF_MAX) + +РЕНОРМАЛИЗАЦИЯ ВЕСОВ (критично): деградированные/None-входы выпадают И из числителя, +И из суммы весов. Доступные веса делятся на свою сумму (renorm) → пропуск данных НЕ +сжимает коэффициент к 1.0 искусственно (иначе 4 из 8 факторов = None всегда тянули +бы coef к нейтрали и недооценивали реальный макрорежим). Все входы None → coef=1.0 +честно (нейтрально, нет сигнала), confidence='low'. + +СЕГМЕНТНЫЕ МОДИФИКАТОРЫ (§9.5 «особенно X»): rate-driven sub-factors (f_rate + +f_mortgage_rate — оба про стоимость заёмных денег) масштабируются под профиль +сегмента. Крупный / дорогой / высокочековый / инвестиционный формат реагирует +КРУЧЕ-негативно на rate↑ (ипотечно-чувствителен); семейный / компактный / ликвидный +comfort/comfort+ / быстрой нарезки — КРУЧЕ-позитивно на rate↓ (+ господдержка). +Реализовано простым набором правил с именованными константами-крутизнами (НЕ фит). + +BACKED vs DEGRADED входы (сверено с полями MonthlyMacro PR2): + • BACKED : f_rate (key_rate), f_mortgage_rate (mortgage_rate_weighted), + f_issuance (mortgage_issued_count + _volume), f_overdue + (mortgage_overdue / mortgage_debt). + • DEGRADED : gov-support / income / cpi / confidence — этих рядов в MonthlyMacro + НЕТ (income/cpi не backfilled, см. macro_series docstring; gov-support + и consumer-confidence вообще не собираются). Они зарезервированы в + формуле/весах для будущего, но СЕЙЧАС всегда contribute None → + выпадают из renorm. Помечаем их явно unavailable, не молчаливый 0. + +Graceful-on-thin-data (дух market_metrics.py / rate_sensitivity.py): пустой макро-ряд +/ все входы None → coefficient=1.0 (нейтрально), confidence='low', НЕ crash. +Детерминированно. numpy НЕ нужен (чистая арифметика). +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Any, Literal + +from sqlalchemy.orm import Session + +from app.services.forecasting.macro_series import ( + MonthlyMacro, + get_monthly_macro, + is_confounded_window, +) + +logger = logging.getLogger(__name__) + +Confidence = Literal["high", "medium", "low"] + +# ── Named-константы: окно / клэмп ───────────────────────────────────────────── + +# Глубина макро-ряда (месяцев назад). 12 мес = «макрорежим за последний год»: §9.5 +# поправляет прогноз на ТЕКУЩИЙ режим, а не на 4-летний цикл (в отличие от §9.6 +# регрессии, которой нужен длинный ряд). Короче окно → свежее «фото» режима. +_DEFAULT_MONTHS_BACK: int = 12 + +# Клэмп-полоса множителя. Центр 1.0; асимметрична намеренно: вниз сильнее (0.6 = +# режим может срезать прогноз спроса до −40%, как в острый rate-шок), вверх мягче +# (1.3 = +30% максимум, господдержка/смягчение не «разгоняют» спрос так же резко, +# как ужесточение его душит — асимметрия рынка ИЖК). Эвристика, уточняется бэктестом. +_MACRO_COEF_MIN: float = 0.6 +_MACRO_COEF_MAX: float = 1.3 +_MACRO_COEF_NEUTRAL: float = 1.0 + +# Окно (мес) для тренда/дельты внутри ряда: сравниваем последнее значение со +# значением N месяцев назад. 6 ≈ полугодие — гасит месячный шум, но ловит разворот +# режима (зеркалит дух лагов §9.6, где полугодовой лаг ловит ипотечный эффект). +_TREND_WINDOW_MONTHS: int = 6 + +# ── Named-константы: веса sub-factors (СУММА backed-весов = 0.45) ────────────── +# Веса — экспертная оценка вклада каждого канала в макрорежим спроса (НЕ фит). +# Заданы в ИСХОДНОМ (полном) наборе из 8 каналов; renorm делит на сумму ДОСТУПНЫХ. +# Backed-каналы (rate/mortgage_rate/issuance/overdue) несут основную массу: ставка и +# стоимость/доступность ипотеки — доминирующий драйвер первичного спроса в РФ. +# Degraded-каналы (gov/income/cpi/confidence) имеют НЕнулевые веса в схеме (резерв +# под будущие ряды), но СЕЙЧАС всегда None → в renorm не попадают. +_W_RATE: float = 0.18 # ключевая ставка — главный канал (стоимость денег) +_W_MORTG_RATE: float = 0.12 # ИЖК-ставка — прямой канал доступности ипотеки +_W_ISSUANCE: float = 0.10 # объём/число выдач ИЖК — фактический спрос-прокси +_W_OVERDUE: float = 0.05 # просрочка/долг — стресс заёмщиков (медленный канал) +_W_GOVSUPPORT: float = 0.20 # DEGRADED (нет ряда) — крупный вес-резерв (льготы) +_W_INCOME: float = 0.12 # DEGRADED — реальные доходы +_W_INFLATION: float = 0.08 # DEGRADED — ИПЦ +_W_CONFIDENCE: float = 0.15 # DEGRADED — потребит. уверенность + +# Имена sub-factors (ключи breakdown / available / weights). Держим как константы, +# чтобы переименование не рассинхронило веса и breakdown молча. +_F_RATE: str = "rate" +_F_MORTG_RATE: str = "mortgage_rate" +_F_ISSUANCE: str = "issuance" +_F_OVERDUE: str = "overdue" +_F_GOVSUPPORT: str = "gov_support" +_F_INCOME: str = "income" +_F_INFLATION: str = "inflation" +_F_CONFIDENCE: str = "confidence" + +# Полная схема весов (исходный знаменатель ДО renorm). dict сохраняет порядок → +# детерминированный обход breakdown. +_WEIGHTS: dict[str, float] = { + _F_RATE: _W_RATE, + _F_MORTG_RATE: _W_MORTG_RATE, + _F_ISSUANCE: _W_ISSUANCE, + _F_OVERDUE: _W_OVERDUE, + _F_GOVSUPPORT: _W_GOVSUPPORT, + _F_INCOME: _W_INCOME, + _F_INFLATION: _W_INFLATION, + _F_CONFIDENCE: _W_CONFIDENCE, +} + +# Каналы, для которых СЕЙЧАС нет ряда в MonthlyMacro → всегда contribute None. +# Зарезервированы в _WEIGHTS под будущие данные; помечаем unavailable явно. +_DEGRADED_FACTORS: frozenset[str] = frozenset( + {_F_GOVSUPPORT, _F_INCOME, _F_INFLATION, _F_CONFIDENCE} +) + +# Rate-driven каналы — к ним применяется сегментный модификатор крутизны (оба про +# стоимость заёмных денег: ставка ЦБ + ИЖК-ставка). +_RATE_DRIVEN_FACTORS: frozenset[str] = frozenset({_F_RATE, _F_MORTG_RATE}) + +# ── Named-константы: нормировка sub-factors (сырое Δ → нудж в [-1,1]) ────────── +# Каждый sub-factor берёт сырую величину/дельту и сжимает в [-1, 1] делением на +# «полную шкалу» (величина, при которой канал считаем максимально нагруженным) с +# клэмпом. Шкалы — эмпирика РФ-рынка 2022-2024; эвристика, уточняется бэктестом. + +# Полная шкала тренда ключевой ставки (п.п. за окно). 8 п.п. ≈ «экстремальный» +# разворот ДКП (напр. 7.5→20 в 2022 ушёл далеко за шкалу → нудж клэмпится в −1). +_RATE_TREND_FULL_PP: float = 8.0 + +# Полная шкала тренда ИЖК-ставки (п.п. за окно). 5 п.п. — резкое движение +# средневзвешенной ИЖК-ставки (она инертнее ключевой, ходит меньшими шагами). +_MORTG_RATE_TREND_FULL_PP: float = 5.0 + +# Полная шкала относительного изменения выдач ИЖК за окно (доля). 0.5 = выдачи +# упали/выросли вдвое (count или volume) → максимальный нудж. Берём СРЕДНЕЕ из +# относительных Δ count и Δ volume (если оба есть; иначе доступный из двух). +_ISSUANCE_REL_FULL: float = 0.5 + +# Полная шкала доли просрочки (overdue/debt). 0.05 = 5% портфеля просрочено — +# уже стрессовый уровень для ИЖК (исторически ИЖК-просрочка в РФ <1-2%). Это +# УРОВЕНЬ (не дельта): высокая доля просрочки = плохой режим → негативный нудж. +_OVERDUE_RATIO_FULL: float = 0.05 +# «Нейтральная» доля просрочки: ниже неё канал не давит (здоровый портфель). 1%. +_OVERDUE_RATIO_NEUTRAL: float = 0.01 + +# ── Named-константы: сегментные крутизны (множители rate-driven нуджей) ──────── +# Базовая крутизна 1.0 (нейтральный сегмент). Чувствительные форматы получают +# >1.0 (круче реагируют на канал стоимости денег — в ОБЕ стороны: сильнее падают +# при rate↑ и сильнее растут при rate↓). Эвристика «особенно X» §9.5. +_STEEP_BASE: float = 1.0 +# Дорогой / бизнес-премиум / высокочековый: ипотека — меньшая доля сделки, но +# инвест-мотив и крупный чек делают спрос капризнее к ставке. +40%. +_STEEP_EXPENSIVE: float = 1.4 +# Крупный формат (3-к / 80+ м²): большой чек → почти всегда ипотека/длинный +# горизонт → сильнее давит ставка. +30%. +_STEEP_LARGE: float = 1.3 +# Инвестиционный мотив (покупка ради доходности, не для жизни): максимально +# чувствителен к стоимости денег (альтернатива — депозит под высокую ставку). +50%. +_STEEP_INVESTMENT: float = 1.5 +# Семейный / компактный / ликвидный comfort: массовый ипотечный спрос, который +# СИЛЬНЕЕ оживает при смягчении + господдержке (адресат льготных программ). +25%. +_STEEP_FAVORED: float = 1.25 +# Потолок суммарной крутизны (профиль может совпасть по нескольким признакам — +# не даём множителям перемножиться в абсурд). Клэмпим итог в [мин, макс]. +_STEEP_MIN: float = 0.7 +_STEEP_MAX: float = 1.8 + +# ── Named-константы: confidence ─────────────────────────────────────────────── +# Confidence падает, когда мало backed-входов доступно ИЛИ окно конфаундено. +# 'high' — доступны ВСЕ 4 backed-канала и окно не шок. +# 'medium' — доступно ≥ _CONF_MEDIUM_MIN_BACKED backed-каналов (частичный сигнал). +# 'low' — почти нет сигнала (0-1 канал) ИЛИ деградация ниже medium-порога. +_CONF_HIGH_MIN_BACKED: int = 4 # все backed-каналы +_CONF_MEDIUM_MIN_BACKED: int = 2 # хотя бы половина backed-каналов + + +@dataclass(frozen=True) +class MacroCoefficient: + """§9.5 макроэкономический коэффициент — композитный множитель спроса. + + Детерминированный результат. `coefficient` — клэмпнутый множитель в + [_MACRO_COEF_MIN, _MACRO_COEF_MAX], центр 1.0. `breakdown` — знаковый вклад + КАЖДОГО sub-factor В ИТОГ (после веса+renorm+сегмент-модификатора) ИЛИ None, + если канал недоступен (явно None, НИКОГДА 0-как-заглушка — пропуск ≠ нулевой + вклад). ADVISORY до валидации — не для production-решений. + """ + + coefficient: float + breakdown: dict[str, float | None] # имя sub-factor → вклад в итог (None=недоступен) + available_inputs: list[str] # backed-каналы, давшие сигнал + unavailable_inputs: list[str] # деградированные / None-каналы + segment_profile: dict[str, Any] + confidence: Confidence + confounded: bool # окно ряда пересекает шок-дату (PR2) + # renorm-множитель весов (доступная_сумма⁻¹·полная_сумма): для отладки/explain. + weight_renorm_factor: float | None = field(default=None) + + def as_dict(self) -> dict[str, Any]: + return { + "coefficient": _round_or_none(self.coefficient, 4), + "breakdown": {k: _round_or_none(v, 4) for k, v in self.breakdown.items()}, + "available_inputs": list(self.available_inputs), + "unavailable_inputs": list(self.unavailable_inputs), + "segment_profile": dict(self.segment_profile), + "confidence": self.confidence, + "confounded": self.confounded, + "weight_renorm_factor": _round_or_none(self.weight_renorm_factor, 4), + } + + +def _round_or_none(value: float | None, digits: int) -> float | None: + return round(value, digits) if value is not None else None + + +# ────────────────────────────────────────────────────────────────────────────── +# Pure-логика — без БД, полностью юнит-тестируемо (чистая арифметика). +# ────────────────────────────────────────────────────────────────────────────── + + +def _clamp(value: float, lo: float, hi: float) -> float: + """Зажать value в [lo, hi]. PURE.""" + return max(lo, min(hi, value)) + + +def f_rate(rate_trend_pp: float | None) -> float | None: + """Нудж от тренда ключевой ставки: rate↑ → негатив. None → недоступен. + + rate_trend_pp = Δ(key_rate) за окно в п.п. (последнее − N мес назад). Растущая + ставка душит спрос → нудж < 0; падающая → нудж > 0. Сжимаем в [-1, 1] делением + на _RATE_TREND_FULL_PP с клэмпом (−знак: рост → отрицательный нудж). PURE. + + Args: + rate_trend_pp: изменение ключевой ставки за окно (п.п.); None = нет данных. + + Returns: + Нудж в [-1, 1] или None (вход None → канал недоступен). + """ + if rate_trend_pp is None: + return None + return _clamp(-rate_trend_pp / _RATE_TREND_FULL_PP, -1.0, 1.0) + + +def f_mortgage_rate(mortgage_rate_trend_pp: float | None) -> float | None: + """Нудж от тренда средневзвешенной ИЖК-ставки: ↑ → негатив. None → недоступен. + + Прямой канал доступности ипотеки: рост ИЖК-ставки удорожает кредит → спрос ↓ → + нудж < 0. Сжимаем в [-1, 1] (_MORTG_RATE_TREND_FULL_PP). PURE. + + Args: + mortgage_rate_trend_pp: Δ mortgage_rate_weighted за окно (п.п.); None = нет. + + Returns: + Нудж в [-1, 1] или None. + """ + if mortgage_rate_trend_pp is None: + return None + return _clamp(-mortgage_rate_trend_pp / _MORTG_RATE_TREND_FULL_PP, -1.0, 1.0) + + +def f_issuance(count_rel_change: float | None, volume_rel_change: float | None) -> float | None: + """Нудж от динамики выдач ИЖК: выдачи↓ → негатив. Оба None → недоступен. + + Относительное изменение за окно: rel = (последнее − база) / |база|. Падение + выдач (count и/или volume) = сжатие фактического спроса → нудж < 0; рост → > 0. + Берём СРЕДНЕЕ из доступных rel-Δ (count, volume); если доступен лишь один — + его. Сжимаем в [-1, 1] (_ISSUANCE_REL_FULL). PURE. + + Args: + count_rel_change: относит. Δ числа выдач за окно; None = нет. + volume_rel_change: относит. Δ объёма выдач за окно; None = нет. + + Returns: + Нудж в [-1, 1] или None (оба входа None → канал недоступен). + """ + parts = [v for v in (count_rel_change, volume_rel_change) if v is not None] + if not parts: + return None + avg_rel = sum(parts) / len(parts) + return _clamp(avg_rel / _ISSUANCE_REL_FULL, -1.0, 1.0) + + +def f_overdue(overdue: float | None, debt: float | None) -> float | None: + """Нудж от доли просрочки ИЖК (overdue/debt): доля↑ → негатив. None → недоступен. + + УРОВЕНЬ (не дельта): высокая доля просрочки в портфеле = стресс заёмщиков = + плохой режим спроса → нудж < 0. Ниже _OVERDUE_RATIO_NEUTRAL (здоровый портфель) + канал не давит (нудж 0). От neutral до _OVERDUE_RATIO_FULL линейно уходит к −1. + Доля ВСЕГДА ≥ 0 → этот канал даёт ТОЛЬКО неположительный нудж (просрочка не + «помогает» спросу). PURE. + + Args: + overdue: просроченная задолженность ИЖК (млн руб); None = нет. + debt: общая задолженность ИЖК (млн руб); None / ≤0 → недоступен (нет базы). + + Returns: + Нудж в [-1, 0] или None (нет данных / нулевой портфель). + """ + if overdue is None or debt is None or debt <= 0: + return None + ratio = overdue / debt + excess = ratio - _OVERDUE_RATIO_NEUTRAL + if excess <= 0: + return 0.0 # здоровый портфель — канал доступен, но нейтрален + span = _OVERDUE_RATIO_FULL - _OVERDUE_RATIO_NEUTRAL + return _clamp(-excess / span, -1.0, 0.0) + + +def _trend_over_window( + values: list[float | None], *, window: int = _TREND_WINDOW_MONTHS +) -> float | None: + """Абсолютный тренд ряда за окно: последнее_known − known_за_window_назад. + + Берём последнюю непустую точку и точку примерно `window` индексов раньше неё + (первую непустую при поиске назад). Любая недоступна → None. Для key_rate / + mortgage_rate возвращает Δ в исходных единицах (п.п.). PURE. + """ + last_idx = _last_known_idx(values) + if last_idx is None: + return None + base_idx = _known_idx_at_or_before(values, last_idx - window) + if base_idx is None or base_idx == last_idx: + return None + last_v = values[last_idx] + base_v = values[base_idx] + if last_v is None or base_v is None: # отсечено выше, но страховка для mypy + return None + return float(last_v) - float(base_v) + + +def _rel_change_over_window( + values: list[float | None], *, window: int = _TREND_WINDOW_MONTHS +) -> float | None: + """Относительный тренд за окно: (последнее − база) / |база|. База ≤0 → None. + + Для count/volume выдач (rel-Δ, безразмерный). Той же логикой поиска последней/ + базовой точки, что _trend_over_window. PURE. + """ + last_idx = _last_known_idx(values) + if last_idx is None: + return None + base_idx = _known_idx_at_or_before(values, last_idx - window) + if base_idx is None or base_idx == last_idx: + return None + last_v = values[last_idx] + base_v = values[base_idx] + if last_v is None or base_v is None or base_v == 0: + return None + return (float(last_v) - float(base_v)) / abs(float(base_v)) + + +def _last_known_idx(values: list[float | None]) -> int | None: + """Индекс последней непустой точки ряда (или None, если все None). PURE.""" + for i in range(len(values) - 1, -1, -1): + if values[i] is not None: + return i + return None + + +def _known_idx_at_or_before(values: list[float | None], idx: int) -> int | None: + """Индекс непустой точки на позиции idx или ближайшей раньше неё. PURE. + + Идём от min(idx, last) назад до первой непустой. idx<0 → None. Так база тренда + устойчива к дыркам ровно на целевой позиции (берём ближайшую известную раньше). + """ + if idx < 0: + return None + start = min(idx, len(values) - 1) + for i in range(start, -1, -1): + if values[i] is not None: + return i + return None + + +def _latest_ratio( + numerators: list[float | None], denominators: list[float | None] +) -> tuple[float | None, float | None]: + """Последняя пара (numerator, denominator) с обоими непустыми значениями. + + Для f_overdue: берём свежайший месяц, где есть И overdue, И debt (доля просрочки + — про текущий УРОВЕНЬ, не тренд). Идём с конца. Нет такой пары → (None, None). + PURE. + """ + n = min(len(numerators), len(denominators)) + for i in range(n - 1, -1, -1): + num = numerators[i] + den = denominators[i] + if num is not None and den is not None: + return num, den + return None, None + + +def segment_steepness(segment_profile: dict[str, Any]) -> float: + """Сегментная крутизна для rate-driven нуджей (§9.5 «особенно X»). PURE. + + Берёт профиль сегмента и возвращает множитель крутизны в [_STEEP_MIN, + _STEEP_MAX], центр _STEEP_BASE=1.0. Чувствительные форматы (дорогой / крупный / + высокочековый / инвестиционный) → >1.0 (круче реагируют на канал стоимости + денег в ОБЕ стороны); favored массовый (семейный / компактный / ликвидный + comfort) → тоже >1.0, но через «оживание при смягчении». + + Признаки профиля (все опциональны; неизвестное → не влияет): + • obj_class : 'бизнес'/'премиум'/'элит…' → expensive; 'комфорт'/'comfort+' + → favored; 'эконом' → favored (массовый ипотечный). + • price_tier : 'high'/'premium'/'бизнес'/'премиум' → expensive; + 'low'/'эконом'/'comfort'… → favored. + • room_bucket : крупная нарезка (3-к / 80+ / '4'/'5+') → large; компактная + (студия / 1-к) → favored. + • is_investment: True → investment (максимальная крутизна). + • is_family / is_compact / is_liquid : True → favored. + + Реализация: стартуем с _STEEP_BASE, домножаем на каждую сработавшую крутизну + (нормированную к base), клэмпим итог. Простой набор правил, НЕ фит. + + Args: + segment_profile: словарь признаков сегмента (любой subset). + + Returns: + Множитель крутизны в [_STEEP_MIN, _STEEP_MAX]. + """ + steep = _STEEP_BASE + for factor in _segment_steepness_factors(segment_profile): + # Домножаем на относительную крутизну (factor/base), чтобы база 1.0 была + # нейтральна, а несколько признаков усиливали друг друга (с потолком). + steep *= factor / _STEEP_BASE + return _clamp(steep, _STEEP_MIN, _STEEP_MAX) + + +def _segment_steepness_factors(profile: dict[str, Any]) -> list[float]: + """Список сработавших крутизн из профиля (порядок детерминирован). PURE.""" + factors: list[float] = [] + + obj_class = _norm_str(profile.get("obj_class")) + price_tier = _norm_str(profile.get("price_tier")) + room_bucket = _norm_str(profile.get("room_bucket")) + + if _is_expensive_class(obj_class) or _is_expensive_tier(price_tier): + factors.append(_STEEP_EXPENSIVE) + if _is_large_room(room_bucket): + factors.append(_STEEP_LARGE) + if _truthy(profile.get("is_investment")): + factors.append(_STEEP_INVESTMENT) + if ( + _is_favored_class(obj_class) + or _is_favored_tier(price_tier) + or _is_compact_room(room_bucket) + or _truthy(profile.get("is_family")) + or _truthy(profile.get("is_compact")) + or _truthy(profile.get("is_liquid")) + ): + factors.append(_STEEP_FAVORED) + return factors + + +def _norm_str(value: Any) -> str | None: + """Привести значение к lowercase-строке (для регистро-независимого матча). PURE.""" + if value is None: + return None + return str(value).strip().lower() + + +def _truthy(value: Any) -> bool: + """True только для явного булева True (None/0/'' → False). PURE.""" + return value is True + + +# Подстроки классов/тиров (lowercase) — РФ-нейминг + латиница. Подстрочный матч, +# чтобы 'премиум-плюс'/'business' и т.п. попадали без точного перечисления. +_EXPENSIVE_CLASS_MARKERS: tuple[str, ...] = ( + "бизнес", + "business", + "премиум", + "premium", + "элит", + "elit", +) +_FAVORED_CLASS_MARKERS: tuple[str, ...] = ( + "комфорт", + "comfort", + "эконом", + "econom", + "стандарт", + "standard", +) +_EXPENSIVE_TIER_MARKERS: tuple[str, ...] = ("high", "premium", "бизнес", "премиум", "верхн") +_FAVORED_TIER_MARKERS: tuple[str, ...] = ( + "low", + "mid", + "эконом", + "комфорт", + "comfort", + "нижн", + "средн", +) +# Крупная нарезка: 3-к и больше / явные крупные метки room_area_bucket / Source A. +_LARGE_ROOM_MARKERS: tuple[str, ...] = ("3-к", "80+", "4-к", "5+", "5-к") +_COMPACT_ROOM_MARKERS: tuple[str, ...] = ("студи", "1-к", "studio") + + +def _has_marker(value: str | None, markers: tuple[str, ...]) -> bool: + """True, если value содержит любую из подстрок-маркеров. PURE.""" + if value is None: + return False + return any(m in value for m in markers) + + +def _is_expensive_class(v: str | None) -> bool: + return _has_marker(v, _EXPENSIVE_CLASS_MARKERS) + + +def _is_favored_class(v: str | None) -> bool: + return _has_marker(v, _FAVORED_CLASS_MARKERS) + + +def _is_expensive_tier(v: str | None) -> bool: + return _has_marker(v, _EXPENSIVE_TIER_MARKERS) + + +def _is_favored_tier(v: str | None) -> bool: + return _has_marker(v, _FAVORED_TIER_MARKERS) + + +def _is_large_room(v: str | None) -> bool: + if v is None: + return False + if v in ("3", "4", "5", "4+", "5+"): # Source A сырые room-ключи + return True + return _has_marker(v, _LARGE_ROOM_MARKERS) + + +def _is_compact_room(v: str | None) -> bool: + if v is None: + return False + if v in ("0", "1", "студия"): # Source A сырые room-ключи (0=студия) + return True + return _has_marker(v, _COMPACT_ROOM_MARKERS) + + +def renormalize_contributions( + nudges: dict[str, float | None], weights: dict[str, float] +) -> tuple[dict[str, float | None], float | None]: + """Свести нуджи sub-factors во взвешенные вклады с РЕНОРМАЛИЗАЦИЕЙ весов. + + Доступные каналы (nudge != None) делят свой вес на СУММУ доступных весов → + итоговые веса доступных суммируются в полную исходную сумму весов (не в ≤1 + из-за выпавших). Так пропуск данных НЕ тянет coef к 1.0 искусственно: каждый + оставшийся канал «забирает» долю выпавших пропорционально своему весу. + + Renorm-множитель = (Σ всех весов) / (Σ доступных весов). Недоступные каналы → + вклад None (явно, не 0). PURE. + + Args: + nudges: имя sub-factor → нудж в [-1,1] или None (недоступен). + weights: имя sub-factor → исходный вес (полная схема). + + Returns: + (contributions, renorm_factor): + • contributions: имя → взвешенный вклад (renorm·вес·нудж) или None. + • renorm_factor: множитель ренормализации или None (нет доступных). + """ + total_weight = sum(weights.values()) + available_weight = sum( + weights.get(name, 0.0) for name, nudge in nudges.items() if nudge is not None + ) + if available_weight <= 0: + # Ни одного доступного канала → все вклады None, renorm неопределён. + return {name: None for name in nudges}, None + + renorm = total_weight / available_weight + contributions: dict[str, float | None] = {} + for name, nudge in nudges.items(): + if nudge is None: + contributions[name] = None + else: + contributions[name] = renorm * weights.get(name, 0.0) * nudge + return contributions, renorm + + +def assemble_coefficient(contributions: dict[str, float | None]) -> float: + """Собрать клэмпнутый множитель из взвешенных вкладов: 1.0 + Σ вкладов. PURE. + + None-вклады (недоступные каналы) пропускаются. Итог клэмпится в + [_MACRO_COEF_MIN, _MACRO_COEF_MAX]. Пустые/все-None вклады → 1.0 (нейтрально). + + Args: + contributions: имя sub-factor → взвешенный вклад или None. + + Returns: + Множитель в [_MACRO_COEF_MIN, _MACRO_COEF_MAX]. + """ + total = _MACRO_COEF_NEUTRAL + sum(c for c in contributions.values() if c is not None) + return _clamp(total, _MACRO_COEF_MIN, _MACRO_COEF_MAX) + + +def _confidence(*, n_backed_available: int, confounded: bool) -> Confidence: + """Confidence §9.5 по числу доступных backed-каналов + шок-окно. PURE. + + 'high' — доступны ВСЕ backed-каналы и окно не конфаундено. + 'medium' — доступно ≥ _CONF_MEDIUM_MIN_BACKED backed-каналов (частичный сигнал) + ЛИБО все backed есть, но окно конфаундено (шок снижает доверие). + 'low' — почти нет сигнала (< medium-порога доступных каналов). + """ + if n_backed_available < _CONF_MEDIUM_MIN_BACKED: + return "low" + if n_backed_available >= _CONF_HIGH_MIN_BACKED and not confounded: + return "high" + return "medium" + + +# ────────────────────────────────────────────────────────────────────────────── +# DB-оркестратор — тонкий, graceful. Pure-логика выше тестируется без него. +# ────────────────────────────────────────────────────────────────────────────── + + +def compute_macro_coefficient( + db: Session, + *, + segment_profile: dict[str, Any] | None = None, + months_back: int = _DEFAULT_MONTHS_BACK, +) -> MacroCoefficient: + """Вычислить §9.5 макроэкономический коэффициент для сегмента. + + ADVISORY (веса — эвристика, не фит): НЕ подключать в production-эндпоинт, пока + не валидировано. Множитель центрирован в 1.0; <1 = режим давит, >1 = поддержит. + + Шаги: + 1. Тянем monthly макро-ряд (PR2). Пустой → coef=1.0, confidence='low'. + 2. Считаем 4 BACKED sub-factors (rate / mortgage_rate / issuance / overdue); + degraded каналы (gov/income/cpi/confidence) → None (нет ряда в MonthlyMacro). + 3. Применяем сегментную крутизну к rate-driven нуджам (rate + mortgage_rate). + 4. Ренормализуем веса по ДОСТУПНЫМ каналам (выпавшие не тянут coef к 1.0). + 5. Собираем clamp(1.0 + Σ вкладов, MIN, MAX); confidence + confounded-флаг. + + Graceful: пусто/все None/ошибка ряда → coef=1.0 (нейтрально), confidence='low', + НЕ crash. Детерминированно. + + Args: + db: SQLAlchemy sync Session. + segment_profile: признаки сегмента (obj_class / price_tier / room_bucket / + is_investment / is_family / is_compact / is_liquid). None → нейтральный. + months_back: глубина макро-ряда (по умолчанию _DEFAULT_MONTHS_BACK). + + Returns: + MacroCoefficient (всегда; coef=1.0 при нехватке данных). + """ + profile = dict(segment_profile or {}) + + macro = get_monthly_macro(db, months_back=months_back) + confounded = _series_confounded(macro) + + # ── 2. Сырые нуджи backed-каналов (degraded → None) ──────────────────────── + base_nudges = _compute_backed_nudges(macro) + + # ── 3. Сегментная крутизна на rate-driven нуджи ──────────────────────────── + steepness = segment_steepness(profile) + nudges = _apply_segment_steepness(base_nudges, steepness) + + # Degraded-каналы добавляем как явные None (для breakdown / unavailable-списка). + for name in _DEGRADED_FACTORS: + nudges.setdefault(name, None) + + # ── 4. Ренормализация весов по доступным каналам ─────────────────────────── + contributions, renorm = renormalize_contributions(nudges, _WEIGHTS) + + # ── 5. Сборка + клэмп + confidence ───────────────────────────────────────── + coefficient = assemble_coefficient(contributions) + + available = [name for name, nudge in nudges.items() if nudge is not None] + unavailable = [name for name, nudge in nudges.items() if nudge is None] + n_backed_available = sum(1 for name in available if name not in _DEGRADED_FACTORS) + confidence = _confidence(n_backed_available=n_backed_available, confounded=confounded) + + logger.info( + "macro_coefficient: coef=%.4f backed_available=%d renorm=%s confounded=%s " + "steepness=%.3f confidence=%s segment=%s", + coefficient, + n_backed_available, + f"{renorm:.4f}" if renorm is not None else None, + confounded, + steepness, + confidence, + profile, + ) + + return MacroCoefficient( + coefficient=coefficient, + breakdown=contributions, + available_inputs=available, + unavailable_inputs=unavailable, + segment_profile=profile, + confidence=confidence, + confounded=confounded, + weight_renorm_factor=renorm, + ) + + +def _compute_backed_nudges(macro: list[MonthlyMacro]) -> dict[str, float | None]: + """Посчитать 4 backed sub-factors из макро-ряда (каждый None при нехватке). PURE-ish. + + Не ходит в БД (получает уже вытянутый ряд) — тестируется напрямую. + + • rate : тренд key_rate за окно → f_rate. + • mortgage_rate : тренд mortgage_rate_weighted за окно → f_mortgage_rate. + • issuance : относит. тренд count & volume выдач → f_issuance. + • overdue : последняя доля overdue/debt → f_overdue. + """ + key_rate = [m.key_rate for m in macro] + mortg_rate = [m.mortgage_rate_weighted for m in macro] + issued_count = [m.mortgage_issued_count for m in macro] + issued_volume = [m.mortgage_issued_volume for m in macro] + overdue_series = [m.mortgage_overdue for m in macro] + debt_series = [m.mortgage_debt for m in macro] + + overdue_last, debt_last = _latest_ratio(overdue_series, debt_series) + + return { + _F_RATE: f_rate(_trend_over_window(key_rate)), + _F_MORTG_RATE: f_mortgage_rate(_trend_over_window(mortg_rate)), + _F_ISSUANCE: f_issuance( + _rel_change_over_window(issued_count), + _rel_change_over_window(issued_volume), + ), + _F_OVERDUE: f_overdue(overdue_last, debt_last), + } + + +def _apply_segment_steepness( + nudges: dict[str, float | None], steepness: float +) -> dict[str, float | None]: + """Домножить rate-driven нуджи (rate + mortgage_rate) на сегментную крутизну. + + После домножения переклэмпиваем в [-1, 1] (крутизна может выгнать за границу). + Не-rate-driven и None-нуджи не трогаем. Возвращает НОВЫЙ dict (не мутирует вход). + PURE. + """ + out: dict[str, float | None] = {} + for name, nudge in nudges.items(): + if nudge is not None and name in _RATE_DRIVEN_FACTORS: + out[name] = _clamp(nudge * steepness, -1.0, 1.0) + else: + out[name] = nudge + return out + + +def _series_confounded(macro: list[MonthlyMacro]) -> bool: + """True, если окно макро-ряда [min..max] пересекает шок-дату (reuse PR2). PURE. + + Пустой ряд → False (нет окна — нечего конфаундить). + """ + if not macro: + return False + months = [m.month for m in macro] + return is_confounded_window(min(months), max(months)) diff --git a/backend/tests/services/forecasting/test_macro_coefficient.py b/backend/tests/services/forecasting/test_macro_coefficient.py new file mode 100644 index 00000000..446fc377 --- /dev/null +++ b/backend/tests/services/forecasting/test_macro_coefficient.py @@ -0,0 +1,548 @@ +"""Unit-тесты §9.5 макроэкономического коэффициента (#951e, ADVISORY). + +Чистые тесты — БЕЗ живой БД (арифметика на синтетике + мок PR2 get_monthly_macro): + • pure sub-factors (f_rate / f_mortgage_rate / f_issuance / f_overdue) — знак + + границы [-1,1] + None-вход → None (недоступен). + • renormalize_contributions — деградированные входы выпадают из числителя И суммы + весов; coef НЕ тянется к 1.0 искусственно; все-None → renorm None. + • segment_steepness — large/expensive/investment → круче (>1.0); favored → >1.0; + нейтральный/неизвестный → 1.0; клэмп. + • assemble_coefficient — клэмп на MIN/MAX; None-вклады пропускаются. + • compute_macro_coefficient (мок PR2): rate↑+issuance↓ → coef<1; rate↓+gov-favored + → coef>1 для favored-сегмента; graceful пусто → 1.0/low. + +ADVISORY-статус (веса — эвристика) проверяется на уровне поведения (центр 1.0, +направленность, renorm, graceful). +""" + +from __future__ import annotations + +import datetime as dt +import math +import os +from unittest.mock import MagicMock, patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services.forecasting.macro_coefficient import ( + _DEGRADED_FACTORS, + _F_ISSUANCE, + _F_MORTG_RATE, + _F_OVERDUE, + _F_RATE, + _MACRO_COEF_MAX, + _MACRO_COEF_MIN, + _MACRO_COEF_NEUTRAL, + _STEEP_BASE, + _STEEP_MAX, + _STEEP_MIN, + _WEIGHTS, + MacroCoefficient, + assemble_coefficient, + compute_macro_coefficient, + f_issuance, + f_mortgage_rate, + f_overdue, + f_rate, + renormalize_contributions, + segment_steepness, +) +from app.services.forecasting.macro_series import MonthlyMacro + +_MACRO = "app.services.forecasting.macro_coefficient.get_monthly_macro" + + +def _months(n: int, *, end: dt.date | None = None) -> list[dt.date]: + """n подряд идущих 1-х чисел месяцев, заканчивая end (по умолчанию 2023-12).""" + end = end or dt.date(2023, 12, 1) + out: list[dt.date] = [] + y, m = end.year, end.month + for _ in range(n): + out.append(dt.date(y, m, 1)) + m -= 1 + if m == 0: + m = 12 + y -= 1 + return list(reversed(out)) + + +def _macro( + months: list[dt.date], + *, + key_rate: list[float | None] | None = None, + mortgage_rate: list[float | None] | None = None, + issued_count: list[float | None] | None = None, + issued_volume: list[float | None] | None = None, + debt: list[float | None] | None = None, + overdue: list[float | None] | None = None, +) -> list[MonthlyMacro]: + """Список MonthlyMacro; невыставленные ряды → все None (degraded-вход).""" + n = len(months) + none_n: list[float | None] = [None] * n + kr = key_rate if key_rate is not None else none_n + mr = mortgage_rate if mortgage_rate is not None else none_n + ic = issued_count if issued_count is not None else none_n + iv = issued_volume if issued_volume is not None else none_n + db_ = debt if debt is not None else none_n + od = overdue if overdue is not None else none_n + out: list[MonthlyMacro] = [] + for i, month in enumerate(months): + out.append( + MonthlyMacro( + month=month, + key_rate=kr[i], + mortgage_rate_weighted=mr[i], + mortgage_issued_count=ic[i], + mortgage_issued_volume=iv[i], + mortgage_debt=db_[i], + mortgage_overdue=od[i], + ) + ) + return out + + +# ── pure: f_rate ────────────────────────────────────────────────────────────── + + +class TestFRate: + def test_rate_up_is_negative(self) -> None: + assert f_rate(4.0) is not None + assert f_rate(4.0) < 0 # ставка ↑ → давит спрос + + def test_rate_down_is_positive(self) -> None: + v = f_rate(-4.0) + assert v is not None and v > 0 # ставка ↓ → поддержит спрос + + def test_zero_trend_is_zero(self) -> None: + assert f_rate(0.0) == 0.0 + + def test_bounded_below_minus_one(self) -> None: + # Экстремальный рост (12 п.п. > full scale 8) → клэмп в −1. + assert f_rate(100.0) == -1.0 + + def test_bounded_above_plus_one(self) -> None: + assert f_rate(-100.0) == 1.0 + + def test_none_is_unavailable(self) -> None: + assert f_rate(None) is None + + +# ── pure: f_mortgage_rate ───────────────────────────────────────────────────── + + +class TestFMortgageRate: + def test_up_negative_down_positive(self) -> None: + up = f_mortgage_rate(3.0) + down = f_mortgage_rate(-3.0) + assert up is not None and up < 0 + assert down is not None and down > 0 + + def test_bounds(self) -> None: + assert f_mortgage_rate(50.0) == -1.0 + assert f_mortgage_rate(-50.0) == 1.0 + + def test_none_unavailable(self) -> None: + assert f_mortgage_rate(None) is None + + +# ── pure: f_issuance ────────────────────────────────────────────────────────── + + +class TestFIssuance: + def test_drop_is_negative(self) -> None: + v = f_issuance(-0.3, -0.3) # выдачи упали → негатив + assert v is not None and v < 0 + + def test_growth_is_positive(self) -> None: + v = f_issuance(0.3, 0.3) + assert v is not None and v > 0 + + def test_averages_two_inputs(self) -> None: + # count +0.5, volume −0.5 → среднее 0 → нудж 0. + assert f_issuance(0.5, -0.5) == 0.0 + + def test_single_available_input(self) -> None: + # Доступен только volume → берём его (count None не обнуляет канал). + v = f_issuance(None, -0.25) + assert v is not None and v < 0 + + def test_bounds(self) -> None: + assert f_issuance(-5.0, -5.0) == -1.0 + assert f_issuance(5.0, 5.0) == 1.0 + + def test_both_none_unavailable(self) -> None: + assert f_issuance(None, None) is None + + +# ── pure: f_overdue ─────────────────────────────────────────────────────────── + + +class TestFOverdue: + def test_high_ratio_is_negative(self) -> None: + # overdue/debt = 30/1000 = 3% > neutral 1% → негатив. + v = f_overdue(30.0, 1000.0) + assert v is not None and v < 0 + + def test_healthy_portfolio_is_neutral_zero(self) -> None: + # 0.5% < neutral 1% → канал доступен, но нудж 0 (не давит). + assert f_overdue(5.0, 1000.0) == 0.0 + + def test_only_non_positive(self) -> None: + # Просрочка не «помогает» спросу: нудж всегда ≤ 0. + v = f_overdue(80.0, 1000.0) # 8% > full 5% → клэмп −1 + assert v == -1.0 + + def test_none_or_zero_debt_unavailable(self) -> None: + assert f_overdue(None, 1000.0) is None + assert f_overdue(10.0, None) is None + assert f_overdue(10.0, 0.0) is None # нулевой портфель → нет базы + + +# ── pure: segment_steepness ─────────────────────────────────────────────────── + + +class TestSegmentSteepness: + def test_neutral_profile_is_base(self) -> None: + assert segment_steepness({}) == _STEEP_BASE + + def test_unknown_fields_ignored(self) -> None: + assert segment_steepness({"foo": "bar", "obj_class": None}) == _STEEP_BASE + + def test_expensive_class_steeper(self) -> None: + assert segment_steepness({"obj_class": "Бизнес"}) > _STEEP_BASE + + def test_premium_tier_steeper(self) -> None: + assert segment_steepness({"price_tier": "premium"}) > _STEEP_BASE + + def test_large_room_steeper(self) -> None: + assert segment_steepness({"room_bucket": "3-к 60-80"}) > _STEEP_BASE + assert segment_steepness({"room_bucket": "4"}) > _STEEP_BASE # Source A ключ + + def test_investment_steeper(self) -> None: + assert segment_steepness({"is_investment": True}) > _STEEP_BASE + + def test_favored_family_compact_steeper(self) -> None: + assert segment_steepness({"obj_class": "комфорт"}) > _STEEP_BASE + assert segment_steepness({"is_family": True}) > _STEEP_BASE + assert segment_steepness({"room_bucket": "студия"}) > _STEEP_BASE + + def test_compound_profile_clamped_to_max(self) -> None: + # Дорогой + крупный + инвестиционный — перемножение крутизн упёрлось бы за + # потолок; клэмп держит в _STEEP_MAX. + steep = segment_steepness( + {"obj_class": "премиум", "room_bucket": "80+ м²", "is_investment": True} + ) + assert steep == _STEEP_MAX + + def test_within_bounds(self) -> None: + for prof in ({}, {"obj_class": "бизнес"}, {"is_investment": True}): + s = segment_steepness(prof) + assert _STEEP_MIN <= s <= _STEEP_MAX + + +# ── pure: renormalize_contributions (РЕНОРМАЛИЗАЦИЯ) ─────────────────────────── + + +class TestRenormalize: + def test_unavailable_dropped_to_none(self) -> None: + nudges: dict[str, float | None] = {"a": -0.5, "b": None, "c": 0.2} + weights = {"a": 0.2, "b": 0.3, "c": 0.1} + contribs, renorm = renormalize_contributions(nudges, weights) + assert contribs["b"] is None # недоступный → None, НЕ 0 + assert contribs["a"] is not None and contribs["c"] is not None + assert renorm is not None + + def test_renorm_factor_scales_by_dropped_weight(self) -> None: + # Доступны a(0.2)+c(0.1)=0.3 из total 0.6 → renorm = 0.6/0.3 = 2.0. + nudges: dict[str, float | None] = {"a": -0.5, "b": None, "c": 0.2} + weights = {"a": 0.2, "b": 0.3, "c": 0.1} + contribs, renorm = renormalize_contributions(nudges, weights) + assert renorm is not None and math.isclose(renorm, 2.0, rel_tol=1e-9) + # Вклад a = renorm·вес·нудж = 2.0·0.2·(−0.5) = −0.2. + assert contribs["a"] is not None + assert math.isclose(contribs["a"], renorm * 0.2 * -0.5, rel_tol=1e-9) + + def test_not_artificially_shrunk_to_neutral(self) -> None: + # Один и тот же сигнал не должен ослабевать оттого, что соседние каналы + # деградировали: renorm возвращает «потерянный» вес оставшимся. + weights = {"a": 0.2, "b": 0.2, "c": 0.2} + # Случай 1: все доступны, все нуджи −1. + all_av: dict[str, float | None] = {"a": -1.0, "b": -1.0, "c": -1.0} + c_all, _ = renormalize_contributions(all_av, weights) + coef_all = assemble_coefficient(c_all) + # Случай 2: только a доступен (нудж −1), b/c деградировали. + one_av: dict[str, float | None] = {"a": -1.0, "b": None, "c": None} + c_one, _ = renormalize_contributions(one_av, weights) + coef_one = assemble_coefficient(c_one) + # Единственный доступный канал с тем же нуджем даёт ТОТ ЖЕ итог (renorm + # компенсирует выпавшие): сигнал не размазан к нейтрали. + assert coef_one == coef_all + + def test_all_unavailable_renorm_none(self) -> None: + nudges: dict[str, float | None] = {"a": None, "b": None} + contribs, renorm = renormalize_contributions(nudges, {"a": 0.2, "b": 0.3}) + assert renorm is None + assert all(v is None for v in contribs.values()) + # Сборка из всех-None вкладов → нейтраль 1.0. + assert assemble_coefficient(contribs) == _MACRO_COEF_NEUTRAL + + def test_available_weights_sum_back_to_total(self) -> None: + # Сумма ЭФФЕКТИВНЫХ весов доступных каналов (renorm·вес) = полной сумме весов + # (это и есть «не сжимать к 1.0»): проверяем на единичных нуджах. + weights = {"a": 0.18, "b": 0.12, "c": 0.10, "d": 0.20} + nudges: dict[str, float | None] = {"a": 1.0, "b": 1.0, "c": None, "d": None} + contribs, renorm = renormalize_contributions(nudges, weights) + eff_sum = sum(v for v in contribs.values() if v is not None) + assert renorm is not None + # eff_sum = renorm·(0.18+0.12)·1.0 = total(0.60). Доступные «забрали» c+d. + assert abs(eff_sum - sum(weights.values())) < 1e-12 + + +# ── pure: assemble_coefficient (клэмп) ──────────────────────────────────────── + + +class TestAssemble: + def test_centered_at_one(self) -> None: + assert assemble_coefficient({}) == _MACRO_COEF_NEUTRAL + assert assemble_coefficient({"a": None}) == _MACRO_COEF_NEUTRAL + + def test_positive_contributions_raise(self) -> None: + c = assemble_coefficient({"a": 0.1, "b": 0.05}) + assert c == _MACRO_COEF_NEUTRAL + 0.15 + + def test_clamped_at_min(self) -> None: + assert assemble_coefficient({"a": -5.0}) == _MACRO_COEF_MIN + + def test_clamped_at_max(self) -> None: + assert assemble_coefficient({"a": 5.0}) == _MACRO_COEF_MAX + + def test_none_contributions_skipped(self) -> None: + assert assemble_coefficient({"a": 0.1, "b": None, "c": -0.05}) == ( + _MACRO_COEF_NEUTRAL + 0.05 + ) + + +# ── compute_macro_coefficient (мок PR2) ─────────────────────────────────────── + + +class TestComputeMacroCoefficient: + def test_rate_up_issuance_down_coef_below_one(self) -> None: + # Ставка растёт + выдачи падают → давящий режим → coef < 1. + n = 12 + months = _months(n) + key_rate = [10.0 + i * 0.5 for i in range(n)] # тренд +5.5 п.п. за окно + issued_count = [10000.0 - i * 400 for i in range(n)] # выдачи падают + issued_volume = [50000.0 - i * 2000 for i in range(n)] + macro = _macro( + months, + key_rate=key_rate, + issued_count=issued_count, + issued_volume=issued_volume, + ) + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient(MagicMock(), segment_profile={}) + assert isinstance(out, MacroCoefficient) + assert out.coefficient < _MACRO_COEF_NEUTRAL + assert _F_RATE in out.available_inputs + assert _F_ISSUANCE in out.available_inputs + # Backed-каналы без данных (mortgage_rate / overdue) → unavailable, не 0. + assert _F_MORTG_RATE in out.unavailable_inputs + assert out.breakdown[_F_MORTG_RATE] is None + + def test_rate_down_favored_segment_coef_above_one(self) -> None: + # Ставка падает + выдачи растут → поддерживающий режим → coef > 1, и для + # favored-сегмента (семейный/компакт) подъём КРУЧЕ (сегмент-модификатор). + n = 12 + months = _months(n) + key_rate = [18.0 - i * 0.5 for i in range(n)] # тренд −5.5 п.п. + issued_count = [8000.0 + i * 400 for i in range(n)] # выдачи растут + issued_volume = [40000.0 + i * 2000 for i in range(n)] + macro = _macro( + months, + key_rate=key_rate, + issued_count=issued_count, + issued_volume=issued_volume, + ) + with patch(_MACRO, return_value=macro): + neutral = compute_macro_coefficient(MagicMock(), segment_profile={}) + with patch(_MACRO, return_value=macro): + favored = compute_macro_coefficient( + MagicMock(), segment_profile={"is_family": True, "room_bucket": "1-к 30-45"} + ) + assert neutral.coefficient > _MACRO_COEF_NEUTRAL + assert favored.coefficient > _MACRO_COEF_NEUTRAL + # Favored реагирует круче на rate↓ → coef ВЫШЕ нейтрального (но клэмп может + # уравнять, если оба упёрлись в MAX — допускаем ≥). + assert favored.coefficient >= neutral.coefficient + + def test_expensive_segment_steeper_negative_on_rate_up(self) -> None: + # Ставка растёт: дорогой/крупный сегмент должен дать coef НИЖЕ нейтрального + # (круче негатив на rate↑). + n = 12 + months = _months(n) + key_rate = [9.0 + i * 0.4 for i in range(n)] + macro = _macro(months, key_rate=key_rate) + with patch(_MACRO, return_value=macro): + neutral = compute_macro_coefficient(MagicMock(), segment_profile={}) + with patch(_MACRO, return_value=macro): + expensive = compute_macro_coefficient( + MagicMock(), + segment_profile={"obj_class": "премиум", "is_investment": True}, + ) + assert expensive.coefficient < neutral.coefficient + + def test_all_backed_available_with_clean_window_high(self) -> None: + # Все 4 backed-канала есть + окно без шок-дат → confidence='high'. + n = 12 + months = _months(n, end=dt.date(2023, 12, 1)) # 2023 без шок-дат PR2 + key_rate = [9.0 + i * 0.2 for i in range(n)] + mortgage_rate = [11.0 + i * 0.15 for i in range(n)] + issued_count = [9000.0 - i * 100 for i in range(n)] + issued_volume = [45000.0 - i * 500 for i in range(n)] + debt = [2_000_000.0 + i * 1000 for i in range(n)] + overdue = [25_000.0 + i * 200 for i in range(n)] + macro = _macro( + months, + key_rate=key_rate, + mortgage_rate=mortgage_rate, + issued_count=issued_count, + issued_volume=issued_volume, + debt=debt, + overdue=overdue, + ) + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient(MagicMock(), segment_profile={}) + assert out.confidence == "high" + assert out.confounded is False + # Все backed доступны; все degraded — нет. + for name in (_F_RATE, _F_MORTG_RATE, _F_ISSUANCE, _F_OVERDUE): + assert name in out.available_inputs + for name in _DEGRADED_FACTORS: + assert name in out.unavailable_inputs + + def test_confounded_window_caps_confidence(self) -> None: + # Окно пересекает шок 2022-02 → confounded=True; даже все backed → не 'high'. + n = 12 + months = _months(n, end=dt.date(2022, 6, 1)) # охватывает 2022-02 + assert any(m == dt.date(2022, 2, 1) for m in months) + key_rate = [9.0 + i * 0.5 for i in range(n)] + mortgage_rate = [11.0 + i * 0.3 for i in range(n)] + issued_count = [9000.0 - i * 100 for i in range(n)] + issued_volume = [45000.0 - i * 500 for i in range(n)] + debt = [2_000_000.0 for _ in range(n)] + overdue = [25_000.0 + i * 300 for i in range(n)] + macro = _macro( + months, + key_rate=key_rate, + mortgage_rate=mortgage_rate, + issued_count=issued_count, + issued_volume=issued_volume, + debt=debt, + overdue=overdue, + ) + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient(MagicMock(), segment_profile={}) + assert out.confounded is True + assert out.confidence != "high" + + def test_partial_backed_is_medium(self) -> None: + # Доступны 2 backed-канала (rate + issuance) → medium (частичный сигнал). + n = 12 + months = _months(n) + key_rate = [9.0 + i * 0.3 for i in range(n)] + issued_count = [9000.0 - i * 100 for i in range(n)] + issued_volume = [45000.0 - i * 500 for i in range(n)] + macro = _macro( + months, + key_rate=key_rate, + issued_count=issued_count, + issued_volume=issued_volume, + ) + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient(MagicMock(), segment_profile={}) + assert out.confidence == "medium" + + def test_graceful_empty_is_neutral_low(self) -> None: + # Пустой макро-ряд → coef=1.0 (нейтрально), confidence='low', не crash. + with patch(_MACRO, return_value=[]): + out = compute_macro_coefficient(MagicMock(), segment_profile={}) + assert out.coefficient == _MACRO_COEF_NEUTRAL + assert out.confidence == "low" + assert out.confounded is False + assert out.weight_renorm_factor is None + # Все каналы недоступны → breakdown целиком None. + assert all(v is None for v in out.breakdown.values()) + + def test_all_none_series_is_neutral_low(self) -> None: + # Сетка месяцев есть, но все ряды None → coef=1.0, low (как degraded). + n = 12 + months = _months(n) + macro = _macro(months) # все ряды None + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient(MagicMock(), segment_profile={}) + assert out.coefficient == _MACRO_COEF_NEUTRAL + assert out.confidence == "low" + + def test_none_segment_profile_defaults_neutral(self) -> None: + # segment_profile=None → нейтральный профиль, не crash. + n = 12 + months = _months(n) + key_rate = [9.0 + i * 0.3 for i in range(n)] + macro = _macro(months, key_rate=key_rate) + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient(MagicMock(), segment_profile=None) + assert isinstance(out, MacroCoefficient) + assert out.segment_profile == {} + + def test_coefficient_always_within_band(self) -> None: + # Любой режим → coef в [MIN, MAX] (клэмп). + n = 12 + months = _months(n) + # Жёсткий шок: ставка +6 п.п., выдачи рухнули, высокая просрочка. + macro = _macro( + months, + key_rate=[9.0 + i * 0.6 for i in range(n)], + mortgage_rate=[10.0 + i * 0.5 for i in range(n)], + issued_count=[10000.0 - i * 700 for i in range(n)], + issued_volume=[50000.0 - i * 3500 for i in range(n)], + debt=[1_000_000.0 for _ in range(n)], + overdue=[80_000.0 + i * 1000 for i in range(n)], + ) + with patch(_MACRO, return_value=macro): + out = compute_macro_coefficient( + MagicMock(), + segment_profile={"obj_class": "премиум", "is_investment": True}, + ) + assert _MACRO_COEF_MIN <= out.coefficient <= _MACRO_COEF_MAX + + +# ── as_dict ─────────────────────────────────────────────────────────────────── + + +class TestMacroCoefficientAsDict: + def test_serialises_and_rounds(self) -> None: + mc = MacroCoefficient( + coefficient=0.876543, + breakdown={"rate": -0.123456, "income": None}, + available_inputs=["rate"], + unavailable_inputs=["income"], + segment_profile={"obj_class": "бизнес"}, + confidence="medium", + confounded=False, + weight_renorm_factor=1.234567, + ) + d = mc.as_dict() + assert d["coefficient"] == 0.8765 + assert d["breakdown"]["rate"] == -0.1235 + assert d["breakdown"]["income"] is None # None-вклад сохраняется + assert d["weight_renorm_factor"] == 1.2346 + assert d["available_inputs"] == ["rate"] + assert d["confidence"] == "medium" + + def test_weights_schema_covers_all_factors(self) -> None: + # Каждый sub-factor имеет вес; degraded-набор ⊂ схеме весов. + assert _F_RATE in _WEIGHTS + assert _F_MORTG_RATE in _WEIGHTS + assert _F_ISSUANCE in _WEIGHTS + assert _F_OVERDUE in _WEIGHTS + assert _DEGRADED_FACTORS.issubset(set(_WEIGHTS))