lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(forecasting): §9.5 macro coefficient (#951e, advisory)

Browse source 
Deterministic composite demand multiplier (centered 1.0, clamped [0.6,1.3]),
corrects a forecast for the macro regime, directional by market segment.
Heuristic named-constant weights over 4 backed sub-factors (rate / mortgage
rate / issuance / overdue) with weight renormalization over available inputs
(degraded gov/income/cpi/confidence drop out of num AND denom, not silently 0).
Segment-steepness modifier on rate-driven channels (expensive/large/investment
steeper-negative on rate↑; family/compact/liquid steeper-positive on rate↓).
Graceful empty -> 1.0/low. Pure, no LLM, no new SQL (reuses PR2 get_monthly_macro).
ADVISORY: not wired into any endpoint. 50 unit tests (forecasting/ total 191).

PR6/wiring follow-ups: gate lone-survivor renorm leverage under low confidence;
reconsider asymmetric favored-segment steepness (resilient on rate↑).
This commit is contained in:
Light1YT 2026-06-03 11:19:59 +05:00
parent f41e2536d8
commit a5e887ae32
3 changed files with 1347 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
backend/app/services/forecasting/__init__.py
View file

@ -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",
]

778
backend/app/services/forecasting/macro_coefficient.py Normal file
View file

@ -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))

548
backend/tests/services/forecasting/test_macro_coefficient.py Normal file
View file

@ -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))