377 lines
20 KiB
Python
377 lines
20 KiB
Python
"""Monthly макро-ряд + детерминированный классификатор режима ключевой ставки.
|
||
|
||
#951 (Site Finder v2 / «GG-форсайт» ТЗ §9.5/§9.6), EPIC 7 «Чувствительность к
|
||
ключевой ставке», PR B (#951b). Это **data-independent фундамент**: monthly
|
||
макро-таймсерия, классификатор тренда ставки (down/up/stable), лаговые помощники
|
||
и маркер «зашумлённого» окна. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM.
|
||
|
||
Потребители (later PRs, НЕ этот): §9.6 key-rate sensitivity, §9.5 макро-коэффициент.
|
||
|
||
Источник данных — таблица macro_indicator (миграция data/sql/123_macro_indicator.sql),
|
||
читается через тонкий reader app.services.site_finder.macro (переиспользуем):
|
||
• key_rate — region 'rf', ДНЕВНОЙ ряд, source 'cbr' (backfill PR B
|
||
#945, 2019-01-01 → сегодня). Ресэмплим daily → monthly.
|
||
• mortgage_* — region 'sverdl', УЖЕ месячные (obs_date нормализован к
|
||
1-му числу через make_date(...,1)). Берём как есть.
|
||
(income/cpi НЕ backfilled — на них не опираемся.)
|
||
|
||
Соглашение по дате: каждый MonthlyMacro.month — ПЕРВОЕ число месяца (YYYY-MM-01).
|
||
Это зеркалит нормализацию mortgage-рядов в backfill (make_date(year, month, 1)),
|
||
поэтому join «месяц-в-месяц» сводится к равенству obs_date без date_trunc.
|
||
|
||
Graceful-on-thin-data (КРИТИЧНО, дух market_metrics.py): пустая таблица / сбой БД
|
||
→ [] (НЕ crash). Месяцы без точки key_rate → carry-forward последнего известного
|
||
значения; до самой первой известной точки carry невозможен → key_rate=None.
|
||
|
||
psycopg v3 / SQLAlchemy text: bind-параметры ВСЕГДА через CAST(:x AS type) —
|
||
никогда :x::type (парсер psycopg3 даёт SyntaxError на :name::type).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from typing import Any, Literal
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.services.site_finder.macro import get_macro_series
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
Regime = Literal["down", "up", "stable"]
|
||
|
||
# ── Named-константы ───────────────────────────────────────────────────────────
|
||
|
||
# Глубина окна по умолчанию для get_monthly_macro (месяцев назад от текущего).
|
||
# 48 мес ≈ 4 года: покрывает полный цикл ставки ЦБ (мягкая 2020 → жёсткая 2022 →
|
||
# смягчение → новый подъём 2023-2024) — достаточно для лаговых регрессий §9.6.
|
||
_DEFAULT_MONTHS_BACK: int = 48
|
||
|
||
# Порог режима ставки: изменение key_rate ≥ этого числа п.п. за окно считается
|
||
# трендом (down/up); меньше по модулю → 'stable' (шум/удержание). 0.25 п.п. —
|
||
# минимальный «значимый» шаг решения ЦБ (совет директоров двигает ставку шагами
|
||
# кратными 0.25-0.50 п.п.; <0.25 п.п. за месяц = фактически плато).
|
||
_REGIME_EPS_PP: float = 0.25
|
||
|
||
# Окно (в месяцах) для оценки локального тренда ставки в classify_regime: сравниваем
|
||
# текущую ставку со ставкой _REGIME_WINDOW_MONTHS назад. 1 = «месяц-к-месяцу».
|
||
# Берём 1: решения ЦБ дискретны и крупны, лишнее сглаживание размывает точку разворота.
|
||
_REGIME_WINDOW_MONTHS: int = 1
|
||
|
||
# ── Confounder-даты (hardcoded — с обоснованием КАЖДОЙ, см. sql.md «нужен why») ──
|
||
# Окна, ПЕРЕСЕКАЮЩИЕ любую из этих дат, зашумлены: продажи в них двигала не ставка,
|
||
# а разовый политический/эпид-шок → downstream (§9.6) понижает по ним confidence.
|
||
# Формат: (date, краткое «почему этот разрыв конфаундит rate↔sales»).
|
||
_CONFOUNDER_DATES: tuple[tuple[date, str], ...] = (
|
||
(
|
||
date(2020, 3, 1),
|
||
"COVID-19 lockdown + запуск льготной ипотеки 6.5%: обвал, затем всплеск "
|
||
"спроса, не связанный с динамикой ключевой ставки",
|
||
),
|
||
(
|
||
date(2022, 2, 1),
|
||
"Геополитический шок 02.2022 + экстренный подъём ставки ЦБ до 20%: "
|
||
"паническая динамика сделок, разрыв обычной связи rate↔sales",
|
||
),
|
||
(
|
||
date(2024, 7, 1),
|
||
"Сворачивание массовой льготной ипотеки с 01.07.2024: ажиотажный «забег» "
|
||
"в июне + провал спроса после — продажи двигала отмена льготы, не ставка",
|
||
),
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MonthlyMacro:
|
||
"""Одна месячная точка макро-ряда (ТЗ §9.5/§9.6).
|
||
|
||
month — ПЕРВОЕ число месяца (YYYY-MM-01). Любое поле = None при отсутствии
|
||
данных за этот месяц (никогда 0-как-заглушка).
|
||
"""
|
||
|
||
month: date
|
||
key_rate: float | None # ставка ЦБ, действовавшая на конец месяца (% годовых)
|
||
mortgage_rate_weighted: float | None # средневзвешенная ИЖК-ставка, %
|
||
mortgage_issued_count: float | None # выдано ИЖК-кредитов, шт
|
||
mortgage_issued_volume: float | None # объём выданных ИЖК, млн руб
|
||
mortgage_debt: float | None # задолженность по ИЖК, млн руб
|
||
mortgage_overdue: float | None # просроченная задолженность по ИЖК, млн руб
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"month": self.month.isoformat(),
|
||
"key_rate": _round_or_none(self.key_rate, 2),
|
||
"mortgage_rate_weighted": _round_or_none(self.mortgage_rate_weighted, 2),
|
||
"mortgage_issued_count": _round_or_none(self.mortgage_issued_count, 0),
|
||
"mortgage_issued_volume": _round_or_none(self.mortgage_issued_volume, 1),
|
||
"mortgage_debt": _round_or_none(self.mortgage_debt, 1),
|
||
"mortgage_overdue": _round_or_none(self.mortgage_overdue, 1),
|
||
}
|
||
|
||
|
||
def _round_or_none(value: float | None, digits: int) -> float | None:
|
||
return round(value, digits) if value is not None else None
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure-логика — без БД, полностью юнит-тестируемо.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
# Месячные mortgage-поля, которые join-им из macro_indicator (region 'sverdl').
|
||
# indicator_type → имя поля MonthlyMacro. mortgage_term намеренно НЕ тянем
|
||
# (downstream §9.6/§9.5 он не нужен; держим MonthlyMacro узким).
|
||
_MORTGAGE_FIELDS: tuple[tuple[str, str], ...] = (
|
||
("mortgage_rate_weighted", "mortgage_rate_weighted"),
|
||
("mortgage_issued_count", "mortgage_issued_count"),
|
||
("mortgage_issued_volume", "mortgage_issued_volume"),
|
||
("mortgage_debt", "mortgage_debt"),
|
||
("mortgage_overdue", "mortgage_overdue"),
|
||
)
|
||
|
||
|
||
def _month_start(d: date) -> date:
|
||
"""Нормализовать дату к 1-му числу её месяца (YYYY-MM-01)."""
|
||
return date(d.year, d.month, 1)
|
||
|
||
|
||
def _carry_forward(values: list[float | None]) -> list[float | None]:
|
||
"""Заполнить None-дыры последним известным значением (LOCF).
|
||
|
||
Дыры ДО первого известного значения остаются None (carry невозможен).
|
||
Чистая функция: не мутирует вход.
|
||
"""
|
||
out: list[float | None] = []
|
||
last: float | None = None
|
||
for v in values:
|
||
if v is not None:
|
||
last = v
|
||
out.append(last)
|
||
return out
|
||
|
||
|
||
def classify_regime(
|
||
rates: list[float | None],
|
||
*,
|
||
eps_pp: float = _REGIME_EPS_PP,
|
||
window_months: int = _REGIME_WINDOW_MONTHS,
|
||
) -> list[Regime]:
|
||
"""Классифицировать локальный тренд ключевой ставки по месяцам.
|
||
|
||
Для каждого месяца i сравниваем ставку с ней же window_months назад:
|
||
• Δ ≤ -eps_pp → 'down' (смягчение ДКП)
|
||
• Δ ≥ +eps_pp → 'up' (ужесточение ДКП)
|
||
• |Δ| < eps_pp → 'stable' (плато/удержание; шум < шага решения ЦБ)
|
||
|
||
Δ берётся на carry-forward версии ряда (LOCF), поэтому одиночные None-дыры
|
||
не рвут тренд. Если на момент i нет ни текущего значения, ни базы для
|
||
сравнения (оба None — типично в самом начале до первой точки) → 'stable'
|
||
(нейтрально, не выдумываем тренд). PURE, без БД.
|
||
|
||
Args:
|
||
rates: ставка по месяцам (хронологически ASC), None = нет данных.
|
||
eps_pp: порог значимого изменения в п.п. (по умолчанию _REGIME_EPS_PP).
|
||
window_months: за сколько месяцев назад берём базу сравнения (≥1).
|
||
|
||
Returns:
|
||
Список того же размера, что rates: 'down' | 'up' | 'stable' на месяц.
|
||
"""
|
||
w = max(1, window_months)
|
||
filled = _carry_forward(rates)
|
||
out: list[Regime] = []
|
||
for i in range(len(filled)):
|
||
cur = filled[i]
|
||
prev = filled[i - w] if i - w >= 0 else None
|
||
if cur is None or prev is None:
|
||
out.append("stable")
|
||
continue
|
||
delta = cur - prev
|
||
if delta <= -eps_pp:
|
||
out.append("down")
|
||
elif delta >= eps_pp:
|
||
out.append("up")
|
||
else:
|
||
out.append("stable")
|
||
return out
|
||
|
||
|
||
def macro_at_lag(
|
||
series: list[MonthlyMacro], target_month: date, lag_months: int
|
||
) -> MonthlyMacro | None:
|
||
"""Вернуть точку макро-ряда на (target_month − lag_months) или None.
|
||
|
||
Лаги, используемые downstream §9.6: 0/1/2/3/6. target_month нормализуется к
|
||
1-му числу. Если такого месяца нет в series (дыра / выход за диапазон) → None.
|
||
PURE, без БД.
|
||
|
||
Args:
|
||
series: список MonthlyMacro (порядок не важен — ищем по равенству month).
|
||
target_month: целевой месяц (нормализуется к YYYY-MM-01).
|
||
lag_months: на сколько месяцев назад сместиться (≥0).
|
||
|
||
Returns:
|
||
MonthlyMacro искомого месяца либо None.
|
||
"""
|
||
anchor = _month_start(target_month)
|
||
wanted = _shift_months(anchor, -lag_months)
|
||
for m in series:
|
||
if m.month == wanted:
|
||
return m
|
||
return None
|
||
|
||
|
||
def _shift_months(d: date, delta: int) -> date:
|
||
"""Сдвинуть дату-1-е-число на delta месяцев (delta может быть отрицательным)."""
|
||
base = _month_start(d)
|
||
total = (base.year * 12 + (base.month - 1)) + delta
|
||
year, month0 = divmod(total, 12)
|
||
return date(year, month0 + 1, 1)
|
||
|
||
|
||
def is_confounded_window(start: date, end: date) -> bool:
|
||
"""True, если окно [start, end] пересекает известную дату-конфаундер.
|
||
|
||
Эти даты (см. _CONFOUNDER_DATES) — разовые шоки, разрывающие обычную связь
|
||
«ключевая ставка ↔ продажи»; downstream §9.6 по таким окнам понижает
|
||
confidence. Сравнение по 1-му числу месяца (даты-конфаундеры — начала
|
||
месяцев). start>end → нормализуем (меняем местами). PURE, без БД.
|
||
|
||
Args:
|
||
start: начало окна.
|
||
end: конец окна (включительно).
|
||
|
||
Returns:
|
||
True если хотя бы одна confounder-дата попадает в [start, end].
|
||
"""
|
||
lo, hi = (start, end) if start <= end else (end, start)
|
||
lo_m = _month_start(lo)
|
||
hi_m = _month_start(hi)
|
||
return any(lo_m <= cdate <= hi_m for cdate, _why in _CONFOUNDER_DATES)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# DB-reader — тонкий, graceful. Pure-логика выше тестируется без него.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
# Ресэмпл ДНЕВНОГО key_rate (region 'rf') → monthly методом last-known-value
|
||
# (ставка, действовавшая на КОНЕЦ месяца). DISTINCT ON (date_trunc('month', ...))
|
||
# + ORDER BY ... obs_date DESC оставляет по одной — самой поздней — точке месяца.
|
||
# Делаем в SQL, а не в Python: иначе пришлось бы тянуть ~1.9k дневных строк в
|
||
# процесс; DISTINCT ON решает last-value за один set-based проход. Месяц,
|
||
# нормализованный к 1-му числу, сразу join-ится к mortgage-рядам.
|
||
_KEY_RATE_MONTHLY_SQL = text(
|
||
"""
|
||
SELECT DISTINCT ON (date_trunc('month', obs_date))
|
||
CAST(date_trunc('month', obs_date) AS date) AS month,
|
||
value
|
||
FROM macro_indicator
|
||
WHERE indicator_type = CAST(:itype AS text)
|
||
AND region = CAST(:region AS text)
|
||
AND value IS NOT NULL
|
||
AND obs_date >= CAST(:since AS date)
|
||
ORDER BY date_trunc('month', obs_date), obs_date DESC
|
||
"""
|
||
)
|
||
|
||
|
||
def get_monthly_macro(
|
||
db: Session, *, months_back: int = _DEFAULT_MONTHS_BACK
|
||
) -> list[MonthlyMacro]:
|
||
"""Собрать monthly макро-ряд за последние months_back месяцев.
|
||
|
||
key_rate (region 'rf', дневной) ресэмплится daily→monthly как last-known-value
|
||
(ставка на конец месяца). Месячные mortgage-ряды (region 'sverdl') берутся как
|
||
есть и join-ятся по году-месяцу. Дыры key_rate между известными точками
|
||
заполняются carry-forward (LOCF); до первой известной точки key_rate=None.
|
||
|
||
Ряд строится по непрерывной сетке месяцев [start .. текущий] (а НЕ только по
|
||
месяцам, где есть данные), чтобы лаги/режим имели регулярный шаг. Месяц без
|
||
ЛЮБЫХ данных всё равно присутствует (все поля None для него — кроме carry key_rate).
|
||
|
||
Graceful: при сбое БД или пустой таблице key_rate сетка месяцев всё равно
|
||
возвращается, но с None-полями (НЕ crash). Пустой список [] — только если
|
||
сама сетка пуста (months_back < 0).
|
||
|
||
Args:
|
||
db: SQLAlchemy sync Session.
|
||
months_back: глубина ряда в месяцах (по умолчанию _DEFAULT_MONTHS_BACK).
|
||
|
||
Returns:
|
||
Список MonthlyMacro по возрастанию month (по непрерывной сетке);
|
||
[] только при пустой сетке (months_back < 0).
|
||
"""
|
||
# month-bucketing в локальной tz сервера (single-region, как и весь codebase)
|
||
today = date.today()
|
||
start = _shift_months(today, -max(0, months_back))
|
||
grid = _month_grid(start, _month_start(today))
|
||
if not grid:
|
||
return []
|
||
|
||
key_rate_by_month = _query_key_rate_monthly(db, since=start)
|
||
mortgage_by_field = _query_mortgage_monthly(db, since=start)
|
||
|
||
# key_rate: разворачиваем на сетку + carry-forward по сетке (LOCF).
|
||
raw_rates: list[float | None] = [key_rate_by_month.get(m) for m in grid]
|
||
filled_rates = _carry_forward(raw_rates)
|
||
|
||
rows: list[MonthlyMacro] = []
|
||
for idx, month in enumerate(grid):
|
||
kwargs: dict[str, float | None] = {
|
||
field: mortgage_by_field.get(field, {}).get(month) for _itype, field in _MORTGAGE_FIELDS
|
||
}
|
||
rows.append(MonthlyMacro(month=month, key_rate=filled_rates[idx], **kwargs))
|
||
|
||
logger.info(
|
||
"get_monthly_macro: months_back=%d grid=%d key_rate_pts=%d",
|
||
months_back,
|
||
len(grid),
|
||
len(key_rate_by_month),
|
||
)
|
||
return rows
|
||
|
||
|
||
def _month_grid(start: date, end: date) -> list[date]:
|
||
"""Непрерывный список 1-х чисел месяцев от start до end включительно (ASC)."""
|
||
lo = _month_start(start)
|
||
hi = _month_start(end)
|
||
if lo > hi:
|
||
return []
|
||
out: list[date] = []
|
||
cur = lo
|
||
while cur <= hi:
|
||
out.append(cur)
|
||
cur = _shift_months(cur, 1)
|
||
return out
|
||
|
||
|
||
def _query_key_rate_monthly(db: Session, *, since: date) -> dict[date, float]:
|
||
"""Ресэмпл дневного key_rate (region 'rf') → {month1st: value}. Graceful → {}."""
|
||
try:
|
||
rows = db.execute(
|
||
_KEY_RATE_MONTHLY_SQL,
|
||
{"itype": "key_rate", "region": "rf", "since": since},
|
||
).all()
|
||
except Exception:
|
||
logger.exception("get_monthly_macro: key_rate query failed")
|
||
return {}
|
||
return {r[0]: float(r[1]) for r in rows if r[0] is not None and r[1] is not None}
|
||
|
||
|
||
def _query_mortgage_monthly(db: Session, *, since: date) -> dict[str, dict[date, float]]:
|
||
"""Месячные mortgage-ряды (region 'sverdl') через reuse get_macro_series.
|
||
|
||
Возвращает {field: {month1st: value}}. obs_date уже нормализован к 1-му числу
|
||
в backfill, но _month_start применяем повторно (страховка). Сбой одного ряда
|
||
не валит остальные (graceful: пустой подсловарь).
|
||
"""
|
||
out: dict[str, dict[date, float]] = {}
|
||
for indicator_type, field in _MORTGAGE_FIELDS:
|
||
try:
|
||
series = get_macro_series(db, indicator_type, region="sverdl", since=since)
|
||
except Exception:
|
||
logger.exception("get_monthly_macro: mortgage series %s failed", indicator_type)
|
||
series = []
|
||
out[field] = {_month_start(d): float(v) for d, v in series}
|
||
return out
|