Light1YT
78b667b018
EPIC 8 advisory deliverables over the #980 demand/supply engine, deterministic, no LLM, no new SQL: - what_to_build.rank_segments: runs compute_demand_supply_forecast across a class×room_bucket[×price] grid (default 3×5×1=15) per district/cad_num/horizon, drops deficit_index=None cells, ranks deficit_index DESC (tie-break confidence DESC then stable segment key). +deficit=build-signal, −=затоварка. - affordability.compute_affordability: §7.9 MAI degraded payment-burden proxy via subsidized weighted rate (_current_mortgage_rate) + annuity on 50m²/240mo. payment_to_income always None (#946 no income), rate_kind always subsidized_weighted (market ~20% absent), confidence always low. - Pure _rank_key / _annuity (rate≤0 → principal/months guard) unit-tested. 62 new tests; 368 forecasting tests green. Advisory + un-wired.
283 lines
16 KiB
Python
283 lines
16 KiB
Python
"""§7.9 Mortgage Affordability Index (MAI) — ДЕГРАДИРОВАННЫЙ прокси платёжной нагрузки.
|
||
|
||
#981 (952-B, Site Finder v2 / «GG-форсайт» ТЗ §7.9), EPIC 8 «что строить +
|
||
доступность». Это **СБОРОЧНЫЙ слой**: своей §9.x-математики не пересобирает —
|
||
переиспользует helper'ы (`_current_mortgage_rate` §аналитики, `build_sales_series`
|
||
для сегментной цены ₽/м²). Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (чистая аннуитет-арифметика
|
||
+ reuse; своего SQL НЕТ).
|
||
|
||
ADVISORY / LOW-CONFIDENCE (КРИТИЧНО — честность важнее цифры):
|
||
Настоящий §7.9 индекс доступности = «рыночный аннуитет (~20%) ÷ доход
|
||
домохозяйства». Обе входные величины в нашей БД ОТСУТСТВУЮТ:
|
||
1. РЫНОЧНАЯ ставка (~20% по ЦБ Янв 2026) — её серии в cbr_mortgage_series НЕТ.
|
||
Единственная доступная ставка — `_current_mortgage_rate`: СРЕДНЕВЗВЕШЕННАЯ
|
||
С льготами (семейная/IT/ДВ), ~7-8% — она ЗАНИЖАЕТ платёж против рынка.
|
||
2. ДОХОД домохозяйства (#946) — в данных проекта его нет вовсе.
|
||
Поэтому MAI здесь = ПРОКСИ платёжной нагрузки (annuity на эталонную квартиру по
|
||
субсидированной ставке), а НЕ истинный рыночный индекс §7.9: `payment_to_income`
|
||
ВСЕГДА None (нет дохода), `rate_kind='subsidized_weighted'` (не рыночная),
|
||
`confidence='low'` ВСЕГДА (деградация по двум каналам), `degraded_reason`
|
||
объясняет оба пробела. Цифра — для explainability/прототипа, не для решения.
|
||
|
||
Graceful-on-thin-data (дух demand_supply_forecast / market_metrics): нет ставки →
|
||
платёж None; нет цены сегмента → платёж None; ставка ≤0 → аннуитет деградирует к
|
||
principal/months (без деления на ноль). НИКОГДА не crash, НИКОГДА 0-как-заглушка.
|
||
Детерминированно.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import statistics
|
||
from dataclasses import dataclass
|
||
from typing import Any, Literal
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.analytics_queries import _current_mortgage_rate
|
||
from app.services.forecasting.sales_series import (
|
||
SalesSource,
|
||
SegmentSpec,
|
||
build_sales_series,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Named-константы ───────────────────────────────────────────────────────────
|
||
|
||
# Эталонная площадь квартиры (м²) для расчёта «типового» тела кредита. ~50 м² ≈
|
||
# медианная 1-2-комнатная квартира масс-рынка ЕКБ (зеркалит ROOM_AREA_BUCKET 1-к
|
||
# 30-45 / 2-к 45-60). Держим явной константой — индекс сопоставим между сегментами
|
||
# при ФИКСИРОВАННОЙ площади (меняется только цена ₽/м²), а не плавает с метражом.
|
||
_REF_AREA_M2: float = 50.0
|
||
|
||
# Срок ипотеки (мес) для аннуитета. 240 мес = 20 лет — типовой срок ИЖК в РФ
|
||
# (медиана ЦБ ~ 20-25 лет; берём круглые 20 для детерминизма). Tunable.
|
||
_ANNUITY_TERM_MONTHS: int = 240
|
||
|
||
# Метка вида ставки в выводе: всегда субсидированная средневзвешенная (рыночной
|
||
# ~20% в БД нет). Зафиксирована константой — downstream не должен принять её за
|
||
# рыночную ставку §7.9.
|
||
_RATE_KIND_SUBSIDIZED: str = "subsidized_weighted"
|
||
|
||
# Источник сегментной цены ₽/м² по умолчанию: Source B (objective_lots) — несёт
|
||
# per-lot цену для price/room-сегментации (Source A её не даёт). Каведат
|
||
# survivorship bias задокументирован в sales_series; для прокси-индекса приемлемо.
|
||
_PRICE_SOURCE: SalesSource = "objective_lots"
|
||
|
||
# Текст деградации — объясняет ОБА пробела (рыночная ставка + доход), чтобы вывод
|
||
# нельзя было прочитать как истинный §7.9 индекс. Используется в degraded_reason.
|
||
_DEGRADED_REASON: str = (
|
||
"Прокси платёжной нагрузки, НЕ истинный §7.9 индекс доступности: "
|
||
"(1) использована СУБСИДИРОВАННАЯ средневзвешенная ставка ИЖК (~7-8%) — "
|
||
"рыночной ставки без льгот (~20%) в БД нет, платёж занижен против рынка; "
|
||
"(2) доход домохозяйства (#946) недоступен, поэтому payment_to_income не "
|
||
"рассчитывается (всегда None). confidence='low' — деградация по двум каналам."
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MortgageAffordabilityIndex:
|
||
"""§7.9 MAI — ДЕГРАДИРОВАННЫЙ прокси платёжной нагрузки (считается на лету).
|
||
|
||
Все денежные поля = None при недостатке данных (НИКОГДА 0-как-заглушка).
|
||
`payment_to_income` ВСЕГДА None (дохода нет в данных, #946). `rate_kind` ВСЕГДА
|
||
'subsidized_weighted' (рыночной ставки нет). `confidence` ВСЕГДА 'low' (прокси,
|
||
не истинный индекс §7.9). `degraded_reason` объясняет оба пробела. Это
|
||
explainability-прокси, НЕ основание для инвест-решения.
|
||
"""
|
||
|
||
segment: dict[str, str | None]
|
||
monthly_payment_rub: float | None # аннуитет на эталонную квартиру (₽/мес)
|
||
rate_used: float | None # годовая ставка (%) — субсидированная средневзвеш.
|
||
rate_kind: str # ВСЕГДА 'subsidized_weighted'
|
||
rate_period: str | None # период точки ставки (explain)
|
||
ref_area_m2: float # эталонная площадь кредита (м²)
|
||
price_per_m2: float | None # сегментная цена ₽/м² (вход тела кредита)
|
||
annuity_term_months: int # срок ипотеки (мес)
|
||
payment_to_income: float | None # ВСЕГДА None — дохода нет (#946)
|
||
payment_at_scenario: dict[int, float] | None # платёж/горизонт при rate_path
|
||
degraded_reason: str # объясняет оба пробела (ставка + доход)
|
||
confidence: Literal["low"] # ВСЕГДА 'low' (деградация)
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"segment": dict(self.segment),
|
||
"monthly_payment_rub": _round_or_none(self.monthly_payment_rub, 0),
|
||
"rate_used": _round_or_none(self.rate_used, 2),
|
||
"rate_kind": self.rate_kind,
|
||
"rate_period": self.rate_period,
|
||
"ref_area_m2": self.ref_area_m2,
|
||
"price_per_m2": _round_or_none(self.price_per_m2, 0),
|
||
"annuity_term_months": self.annuity_term_months,
|
||
"payment_to_income": self.payment_to_income,
|
||
"payment_at_scenario": (
|
||
{h: round(v, 0) for h, v in self.payment_at_scenario.items()}
|
||
if self.payment_at_scenario is not None
|
||
else None
|
||
),
|
||
"degraded_reason": self.degraded_reason,
|
||
"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 _annuity(
|
||
principal: float | None,
|
||
annual_rate: float | None,
|
||
months: int,
|
||
) -> float | None:
|
||
"""Стандартный аннуитетный платёж по кредиту. PURE, graceful (без /0).
|
||
|
||
Формула (annual_rate в ПРОЦЕНТАХ годовых → месячная доля i = rate/100/12):
|
||
payment = principal · i · (1+i)^n / ((1+i)^n − 1).
|
||
|
||
Правила деградации (честность важнее «красивой» цифры):
|
||
• principal None ИЛИ annual_rate None → None (нет тела/ставки — нечего считать).
|
||
• months ≤ 0 → None (срок не задан).
|
||
• annual_rate ≤ 0 → principal / months (беспроцентный кредит: НЕ делим на
|
||
ноль в формуле, где i=0 даёт 0/0; равномерное погашение тела).
|
||
|
||
Args:
|
||
principal: тело кредита (₽); None → None.
|
||
annual_rate: годовая ставка в процентах (напр. 7.5); None → None; ≤0 →
|
||
беспроцентная деградация principal/months.
|
||
months: срок кредита (мес); ≤0 → None.
|
||
|
||
Returns:
|
||
Месячный платёж (₽) или None.
|
||
"""
|
||
if principal is None or annual_rate is None:
|
||
return None
|
||
if months <= 0:
|
||
return None
|
||
if annual_rate <= 0:
|
||
# Беспроцентный кредит: формула вырождается (i=0 → 0/0). Равномерное тело.
|
||
return principal / float(months)
|
||
i = annual_rate / 100.0 / 12.0
|
||
growth = (1.0 + i) ** months
|
||
return principal * i * growth / (growth - 1.0)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# DB-оркестратор — тонкий, graceful. Pure-логика выше тестируется без него.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def compute_affordability(
|
||
db: Session,
|
||
*,
|
||
spec: SegmentSpec,
|
||
price_per_m2: float | None = None,
|
||
rate_path: dict[int, float] | None = None,
|
||
ref_area_m2: float = _REF_AREA_M2,
|
||
price_source: SalesSource = _PRICE_SOURCE,
|
||
) -> MortgageAffordabilityIndex:
|
||
"""§7.9 MAI — ДЕГРАДИРОВАННЫЙ прокси платёжной нагрузки для сегмента.
|
||
|
||
ADVISORY / LOW-CONFIDENCE: это ПРОКСИ, а НЕ истинный §7.9 индекс доступности.
|
||
Рыночной ставки (~20%) и дохода домохозяйства (#946) в данных нет — см. module
|
||
docstring и `degraded_reason`. `payment_to_income` ВСЕГДА None.
|
||
|
||
Что считает:
|
||
• rate, period = _current_mortgage_rate(db) — субсидированная средневзвеш.
|
||
ставка (~7-8%; ЕДИНСТВЕННАЯ в БД). Это rate_kind='subsidized_weighted'.
|
||
• price_per_m2 — из аргумента ИЛИ сегментная средняя (reuse build_sales_series,
|
||
усреднение НЕнулевых месяцев). None → платёж None (graceful).
|
||
• monthly_payment = _annuity(principal=price×ref_area, annual_rate=rate,
|
||
months=_ANNUITY_TERM_MONTHS).
|
||
• payment_at_scenario[h] = _annuity(... annual_rate=rate_path[h]) — платёж на
|
||
каждом горизонте сценарной ставки (None rate_path → поле None).
|
||
|
||
Graceful: нет ставки/цены → платёж None; ставка ≤0 → аннуитет деградирует к
|
||
principal/months. НИКОГДА не crash. confidence ВСЕГДА 'low'. Детерминированно.
|
||
|
||
Args:
|
||
db: SQLAlchemy sync Session.
|
||
spec: целевой сегмент (для сегментной цены, если price_per_m2 не задан).
|
||
price_per_m2: цена ₽/м² (None → берём сегментную среднюю reuse-ом).
|
||
rate_path: сценарный {horizon: годовая ставка %} для payment_at_scenario;
|
||
None → payment_at_scenario None.
|
||
ref_area_m2: эталонная площадь тела кредита (по умолчанию _REF_AREA_M2).
|
||
price_source: источник сегментной цены (по умолчанию _PRICE_SOURCE = B).
|
||
|
||
Returns:
|
||
MortgageAffordabilityIndex (всегда; тонко → None-поля, confidence='low').
|
||
"""
|
||
segment = spec.as_dict()
|
||
rate, rate_period = _current_mortgage_rate(db)
|
||
|
||
resolved_price = (
|
||
price_per_m2
|
||
if price_per_m2 is not None
|
||
else _segment_avg_price(db, spec=spec, source=price_source)
|
||
)
|
||
principal = resolved_price * ref_area_m2 if resolved_price is not None else None
|
||
|
||
monthly_payment = _annuity(principal, rate, _ANNUITY_TERM_MONTHS)
|
||
|
||
payment_at_scenario: dict[int, float] | None = None
|
||
if rate_path is not None:
|
||
scenario: dict[int, float] = {}
|
||
for horizon, scenario_rate in rate_path.items():
|
||
payment = _annuity(principal, scenario_rate, _ANNUITY_TERM_MONTHS)
|
||
if payment is not None:
|
||
scenario[horizon] = payment
|
||
payment_at_scenario = scenario
|
||
|
||
logger.info(
|
||
"affordability: segment=%s rate=%s period=%s price_per_m2=%s ref_area=%.1f "
|
||
"monthly_payment=%s confidence=low (DEGRADED proxy, payment_to_income=None)",
|
||
segment,
|
||
_round_or_none(rate, 2),
|
||
rate_period,
|
||
_round_or_none(resolved_price, 0),
|
||
ref_area_m2,
|
||
_round_or_none(monthly_payment, 0),
|
||
)
|
||
|
||
return MortgageAffordabilityIndex(
|
||
segment=segment,
|
||
monthly_payment_rub=monthly_payment,
|
||
rate_used=rate,
|
||
rate_kind=_RATE_KIND_SUBSIDIZED,
|
||
rate_period=rate_period,
|
||
ref_area_m2=ref_area_m2,
|
||
price_per_m2=resolved_price,
|
||
annuity_term_months=_ANNUITY_TERM_MONTHS,
|
||
payment_to_income=None, # дохода нет в данных (#946) — НИКОГДА не считаем
|
||
payment_at_scenario=payment_at_scenario,
|
||
degraded_reason=_DEGRADED_REASON,
|
||
confidence="low", # ВСЕГДА low — деградированный прокси, не истинный §7.9
|
||
)
|
||
|
||
|
||
def _segment_avg_price(db: Session, *, spec: SegmentSpec, source: SalesSource) -> float | None:
|
||
"""Сегментная средняя цена ₽/м² — reuse build_sales_series (НЕ новый SQL).
|
||
|
||
Берём monthly ряд продаж сегмента и усредняем НЕнулевые (не-None) месячные
|
||
avg_price_per_m2 — простое среднее по месяцам с реальными сделками. Нет ни
|
||
одного месяца с ценой (тонкий сегмент / сбой) → None (graceful, НЕ 0). PURE-ish
|
||
(тонкая обёртка над reuse).
|
||
|
||
Args:
|
||
db: SQLAlchemy sync Session.
|
||
spec: сегмент рынка.
|
||
source: источник ряда (objective_lots несёт per-lot цену).
|
||
|
||
Returns:
|
||
Средняя цена ₽/м² по месяцам со сделками или None.
|
||
"""
|
||
series = build_sales_series(db, spec=spec, source=source)
|
||
prices = [p for p in series.avg_price_per_m2 if p is not None]
|
||
if not prices:
|
||
return None
|
||
return statistics.fmean(prices)
|