"""§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))