"""§9.4 коэффициент нормализации спроса — дисконт «бумного» темпа под новый режим. #951 (Site Finder v2 / «GG-форсайт» ТЗ §9.4), EPIC 7 «Чувствительность к ключевой ставке», sub-PR 5 (#951f). Проблема §9.4: «Если объект хорошо продавался при льготной ипотеке и низкой ставке, система НЕ должна автоматически переносить этот темп в будущее при высокой ставке. Нужен коэффициент нормализации спроса.» Т.е. наблюдаемый темп продаж был снят при ОДНОМ режиме ставки (напр. низкая ставка + льготная ипотека = «бум»); при проекции в будущее с ДРУГИМ режимом (напр. высокая ставка) этот темп надо ДИСКОНТИРОВАТЬ, а не переносить наивно. Коэффициент `norm` домножается на спроектированный темп: <1 = срезать боомный темп, ≈1 = режимы совпали (нечего корректировать), >1 = осторожный аплифт, если будущее МЯГЧЕ окна наблюдения. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (чистая математика + reuse PR2/PR3, без своего SQL). ADVISORY-СТАТУС: модуль СОВЕТУЮЩИЙ и НЕ подключён ни к одному production-эндпоинту в этом PR (как §9.6 rate_sensitivity / §9.5 macro_coefficient — все advisory до валидации бэктестом PR6). Цифры пригодны для explainability/прототипа, но НЕ для production-решений, пока β из §9.6 не проверен на out-of-sample. ФОРМУЛА (логлинейная, на той же Δln-шкале, что β из §9.6): norm = clamp(exp(beta · (rate_future − rate_window_avg)), _NORM_MIN, _NORM_MAX) ЗНАКОВАЯ ЛОГИКА (почему future > window → ДИСКОНТ): • β — шринкнутый slope §9.6 (PR3) на Δln(продаж) при +1 п.п. ставки. β < 0: продажи ПАДАЮТ, когда ставка растёт (см. rate_sensitivity gate slope<0). • rate_window_avg — средняя ключевая ставка за окно, ИЗ которого снят наблюдаемый темп (период «бума»). rate_future — ожидаемая ставка на горизонте прогноза (аргумент вызывающего). • Δ = rate_future − rate_window_avg. – Будущее ЖЁСТЧЕ окна (future > window, Δ > 0): β·Δ < 0 (β<0·+) → exp(<0) < 1 → ДИСКОНТ. Это и есть §9.4: не тащить низкоставочный бумный темп в высокую ставку. – Режимы совпали (future ≈ window, Δ ≈ 0): exp(≈0) ≈ 1 → темп не трогаем. – Будущее МЯГЧЕ окна (future < window, Δ < 0): β·Δ > 0 → exp(>0) > 1 → аплифт (наблюдали при жёсткой ставке, проектируем в мягкую — темп может вырасти). Клэмп держит коэффициент в разумной полосе даже при экстремальном Δ. ЧЕСТНАЯ ДЕГРАДАЦИЯ (КРИТИЧНО — дух market_metrics.py / rate_sensitivity.py): если β из §9.6 НЕнадёжен (confidence='low': не прошёл gate / шринкнут к ~0 / тонкий сегмент) ИЛИ β недоступен (None) → norm = 1.0 (НЕЙТРАЛЬНО) + applied=False + confidence='low'. Тогда мы НЕ переносим бумный темп наивно (norm не >1), но и НЕ выдумываем дисконт, которому нет статистического основания. Реальную коррекцию (норм ≠ 1.0) применяем ТОЛЬКО когда β из §9.6 заслуживает доверия. Graceful: пустой макро-ряд / нет окна → rate_window_avg=None → norm=1.0, low, не crash. Детерминированно. numpy НЕ нужен (чистая арифметика). """ from __future__ import annotations import logging import math from dataclasses import dataclass from typing import Any, Literal from sqlalchemy.orm import Session from app.services.forecasting.macro_series import MonthlyMacro, get_monthly_macro from app.services.forecasting.regression import compute_rate_regime_sensitivity from app.services.forecasting.sales_series import SegmentSpec logger = logging.getLogger(__name__) Confidence = Literal["high", "medium", "low"] # ── Named-константы ─────────────────────────────────────────────────────────── # Глубина окна (месяцев назад). 48 мес ≈ 4 года — зеркалит _DEFAULT_MONTHS_BACK # PR2/PR3: β §9.6 и средняя ставка окна должны браться из ОДНОГО периода (тот же # историч. ряд, из которого снят наблюдаемый темп продаж). _DEFAULT_MONTHS_BACK: int = 48 # Полоса клэмпа коэффициента нормализации. Центр-нейтраль 1.0 (режимы совпали). # Асимметрична намеренно (как клэмп §9.5 macro_coefficient): вниз шире, вверх уже. # _NORM_MIN = 0.5 — максимальный ДИСКОНТ: даже при резком ужесточении срезаем # бумный темп не более чем вдвое (−50%); ниже = модель уже не «нормализация», # а полное обнуление спроса, чему §9.4-β не даёт основания. # _NORM_MAX = 1.2 — максимальный АПЛИФТ: смягчение режима относительно окна # наблюдения может добавить темпу не более +20%. Жёстче ограничиваем рост, чем # падение: экстраполировать БОЛЬШИЙ спрос в более мягкий режим рискованнее, чем # осторожно срезать (наблюдённый темп — верхняя планка, а не нижняя). Эвристика, # уточняется бэктестом PR6. _NORM_MIN: float = 0.5 _NORM_MAX: float = 1.2 _NORM_NEUTRAL: float = 1.0 @dataclass(frozen=True) class DemandNormalization: """§9.4 коэффициент нормализации спроса (дисконт темпа под смену режима ставки). Детерминированный результат. `coefficient` — клэмпнутый множитель в [_NORM_MIN, _NORM_MAX] для спроектированного темпа продаж (<1 = срезать бумный темп, ≈1 = режимы совпали, >1 = осторожный аплифт). `applied` = False, когда деградировали к нейтрали 1.0 (β §9.6 ненадёжен/недоступен) — честный сигнал, что реальной коррекции НЕ внесено. Числовые поля = None при недостатке данных (никогда 0-как-заглушка). ADVISORY до валидации бэктестом PR6 — не для production-решений. """ coefficient: float # клэмпнутый norm ∈ [_NORM_MIN, _NORM_MAX] beta: float | None # шринкнутый β §9.6 (Δln на +1 п.п.); None если недоступен rate_future: float # ожидаемая ставка на горизонте (аргумент вызывающего) rate_window_avg: float | None # средняя ставка за окно наблюдения темпа, п.п. rate_delta: float | None # rate_future − rate_window_avg (None если avg None) applied: bool # True = внесена реальная коррекция; False = деградация к 1.0 segment: dict[str, str | None] confidence: Confidence def as_dict(self) -> dict[str, Any]: return { "coefficient": _round_or_none(self.coefficient, 4), "beta": _round_or_none(self.beta, 4), "rate_future": _round_or_none(self.rate_future, 2), "rate_window_avg": _round_or_none(self.rate_window_avg, 2), "rate_delta": _round_or_none(self.rate_delta, 2), "applied": self.applied, "segment": dict(self.segment), "confidence": self.confidence, } 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 normalization_factor( beta: float | None, rate_future: float, rate_window_avg: float | None, *, norm_min: float = _NORM_MIN, norm_max: float = _NORM_MAX, ) -> float: """Коэффициент нормализации спроса: clamp(exp(β·Δ), norm_min, norm_max). PURE. Δ = rate_future − rate_window_avg. Знаковая логика (β < 0 в §9.6): • future > window (Δ>0) → β·Δ<0 → exp<1 → ДИСКОНТ (не тащим бумный темп в более жёсткий режим — суть §9.4). • future ≈ window (Δ≈0) → exp≈1 → темп не трогаем (режимы совпали). • future < window (Δ<0) → β·Δ>0 → exp>1 → аплифт (будущее мягче окна). β=None ИЛИ rate_window_avg=None → 1.0 (нейтрально: нет основания корректировать, но и наивного переноса не делаем). Клэмп защищает от экстремального Δ. Без БД. Args: beta: шринкнутый slope §9.6 на Δln(продаж) при +1 п.п. ставки (обычно <0); None = β недоступен/ненадёжен → нейтраль. rate_future: ожидаемая ключевая ставка на горизонте прогноза (п.п.). rate_window_avg: средняя ставка за окно наблюдения темпа (п.п.); None → нейтраль (нет базы сравнения режимов). norm_min: нижняя граница клэмпа (по умолчанию _NORM_MIN). norm_max: верхняя граница клэмпа (по умолчанию _NORM_MAX). Returns: Коэффициент нормализации в [norm_min, norm_max]; ровно 1.0 при β/avg = None. """ if beta is None or rate_window_avg is None: return _NORM_NEUTRAL delta = rate_future - rate_window_avg raw = math.exp(beta * delta) return _clamp(raw, norm_min, norm_max) def _window_avg_rate(macro: list[MonthlyMacro]) -> float | None: """Средняя ключевая ставка по окну макро-ряда (период наблюдения темпа). PURE. Берём среднее всех НЕпустых key_rate в ряду (месяцы без ставки игнорируем — не подмешиваем 0). Это «режим», при котором снят наблюдаемый темп продаж, — база сравнения с rate_future в §9.4. Нет ни одной точки key_rate → None (graceful: окно не определено → нормализацию не применяем). Args: macro: monthly макро-ряд (PR2) за окно наблюдения. Returns: Средняя key_rate (п.п.) или None, если в окне нет ни одной ставки. """ rates = [m.key_rate for m in macro if m.key_rate is not None] if not rates: return None return sum(rates) / len(rates) def _neutral( *, segment: dict[str, str | None], rate_future: float, beta: float | None, rate_window_avg: float | None, ) -> DemandNormalization: """Нейтральный результат (norm=1.0, applied=False, low). PURE. Используется, когда β §9.6 ненадёжен/недоступен ИЛИ окно ставки не определено: не тащим бумный темп наивно (norm не >1), но и не выдумываем дисконт без статистического основания. rate_delta заполняем, если оба конца известны (для explainability), хотя коррекция не внесена. """ rate_delta = rate_future - rate_window_avg if rate_window_avg is not None else None return DemandNormalization( coefficient=_NORM_NEUTRAL, beta=beta, rate_future=rate_future, rate_window_avg=rate_window_avg, rate_delta=rate_delta, applied=False, segment=segment, confidence="low", ) # ────────────────────────────────────────────────────────────────────────────── # DB-оркестратор — тонкий, graceful. Pure-логика выше тестируется без него. # ────────────────────────────────────────────────────────────────────────────── def compute_demand_normalization( db: Session, *, spec: SegmentSpec, rate_future: float, months_back: int = _DEFAULT_MONTHS_BACK, ) -> DemandNormalization: """Вычислить §9.4 коэффициент нормализации спроса для сегмента. ADVISORY (β §9.6 advisory до бэктеста PR6): НЕ подключать в production-эндпоинт. Коэффициент домножается на спроектированный темп продаж: <1 = срезать бумный темп под более жёсткий будущий режим ставки (суть §9.4), ≈1 = режимы совпали, >1 = осторожный аплифт под более мягкое будущее. Шаги: 1. β §9.6 (PR3 compute_rate_sensitivity) для сегмента на том же окне. 2. Средняя ставка окна наблюдения (PR2 get_monthly_macro → _window_avg_rate) — «режим», при котором снят наблюдаемый темп продаж. 3. ГЕЙТ ЧЕСТНОСТИ: β недоступен (None) ИЛИ §9.6 confidence='low' (gate провален / шринк к ~0 / тонкий сегмент) ИЛИ окно ставки не определено → нейтраль (norm=1.0, applied=False, low). Реальную коррекцию вносим только при надёжном β. 4. norm = normalization_factor(β, rate_future, rate_window_avg); applied=True. 5. confidence НЕ выше §9.6 confidence (наследуем доверие к β — коэффициент не надёжнее своего входа). Graceful: пусто/ошибка/тонко → norm=1.0, applied=False, low, не crash. Детерминированно. Args: db: SQLAlchemy sync Session. spec: сегмент рынка, чей наблюдаемый темп нормализуем (PR1 SegmentSpec). rate_future: ожидаемая ключевая ставка на горизонте прогноза (п.п.). months_back: глубина окна наблюдения (по умолчанию _DEFAULT_MONTHS_BACK); передаётся И в §9.6, И в макро-ряд — окна совпадают. Returns: DemandNormalization (всегда; norm=1.0 + applied=False при нехватке данных). """ segment = spec.as_dict() # ── 1. β §9.6 (Almon-ADL long-run multiplier via the validated #978 estimator) sensitivity = compute_rate_regime_sensitivity(db, spec=spec, months_back=months_back) beta = sensitivity.beta # ── 2. Средняя ставка окна наблюдения ────────────────────────────────────── macro = get_monthly_macro(db, months_back=months_back) rate_window_avg = _window_avg_rate(macro) # ── 3. Гейт честности: ненадёжный β / нет окна → нейтраль ─────────────────── if beta is None or sensitivity.confidence == "low" or rate_window_avg is None: logger.info( "demand_normalization: degraded to neutral (segment=%s beta=%s " "sensitivity_conf=%s rate_window_avg=%s) → norm=1.0 applied=False", segment, beta, sensitivity.confidence, rate_window_avg, ) return _neutral( segment=segment, rate_future=rate_future, beta=beta, rate_window_avg=rate_window_avg, ) # ── 4. Реальная коррекция ────────────────────────────────────────────────── coefficient = normalization_factor(beta, rate_future, rate_window_avg) rate_delta = rate_future - rate_window_avg # ── 5. confidence ≤ §9.6 confidence (не надёжнее своего β) ────────────────── confidence = sensitivity.confidence logger.info( "demand_normalization: segment=%s beta=%.4f rate_future=%.2f window_avg=%.2f " "delta=%.2f norm=%.4f confidence=%s applied=True", segment, beta, rate_future, rate_window_avg, rate_delta, coefficient, confidence, ) return DemandNormalization( coefficient=coefficient, beta=beta, rate_future=rate_future, rate_window_avg=rate_window_avg, rate_delta=rate_delta, applied=True, segment=segment, confidence=confidence, )