129 lines
6.4 KiB
Python
129 lines
6.4 KiB
Python
"""Thin deterministic reader for the unified macro_indicator series table.
|
||
|
||
ТЗ §8.8 / #945 (EPIC «Унифицированная сущность Макропоказатель»), PR A.
|
||
|
||
Архитектурный принцип: ДЕТЕРМИНИРОВАННО, БЕЗ LLM, без внешних вызовов — это просто
|
||
типизированное чтение временных рядов из таблицы macro_indicator (создаётся и
|
||
заполняется миграцией data/sql/123_macro_indicator.sql).
|
||
|
||
Это то, что будут потреблять последующие эпики: Mortgage Affordability Index,
|
||
макро-коэффициент, key-rate sensitivity. Держим helper МИНИМАЛЬНЫМ.
|
||
|
||
psycopg v3 / SQLAlchemy text: bind-параметры ВСЕГДА через CAST(:x AS type) —
|
||
никогда :x::type (парсер psycopg3 игнорирует :name::type → SyntaxError).
|
||
|
||
NB: analytics_queries._current_mortgage_rate НЕ трогаем в этом PR (отдельная задача).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
# ── Маппинг cbr_mortgage_series.series_id → indicator_type (ЗЕРКАЛО SQL-миграции) ─
|
||
# Это pure-Python зеркало CASE series_id-выражения из 123_macro_indicator.sql. Сам
|
||
# backfill делается set-based в SQL (миграция не импортирует Python) — здесь дублируем
|
||
# логику ТОЛЬКО для unit-тестируемости и для возможного переиспользования из Python.
|
||
# При изменении маппинга в миграции — синхронно правьте здесь (и наоборот).
|
||
#
|
||
# Ключ — устойчивый series_id (НЕ title!): в prod строка «в рублях» живёт именно в
|
||
# series_id, а title общий и её не содержит, поэтому матчим series_id точным равенством.
|
||
# Берём ТОЛЬКО _в_рублях-варианты семейства _mortgage_ (ИЖК, форма 02): маппинг и
|
||
# _в_рублях, и _итого на один indicator_type коллиднул бы по PK (type,region,obs_date).
|
||
# НЕ включены: _итого-варианты, _housing_ (02_06/07/08), _scpa_mortgage_ (02_15-18),
|
||
# _Share_ (02_44) — все они дают None и в backfill не попадают.
|
||
_CBR_SERIES_MAP: dict[str, tuple[str, str]] = {
|
||
"02_13_Rates_mortgage_ставка_в_рублях": ("mortgage_rate_weighted", "%"),
|
||
"02_13_Rates_mortgage_срок_в_руб": ("mortgage_term", "мес"),
|
||
"02_10_Quantity_mortgage_в_рублях": ("mortgage_issued_count", "шт"),
|
||
"02_11_New_loans_mortgage_в_рублях": ("mortgage_issued_volume", "млн руб"),
|
||
"02_14_Debt_mortgage_в_рублях": ("mortgage_debt", "млн руб"),
|
||
"02_14_Debt_mortgage_в_т.ч._просроч._в_рублях": ("mortgage_overdue", "млн руб"),
|
||
}
|
||
|
||
|
||
def classify_cbr_series(series_id: str | None) -> tuple[str, str] | None:
|
||
"""Pure helper: cbr_mortgage_series.series_id → (indicator_type, unit) | None.
|
||
|
||
Детерминированно, без БД. Ключуем по УСТОЙЧИВОМУ series_id (не по title — title
|
||
в prod общий и не содержит «в рублях»). Возвращает None для рядов, которые НЕ
|
||
сопоставляются ни одному чистому макротипу (они сознательно пропускаются в backfill:
|
||
все _итого-варианты, _housing_, _scpa_mortgage_, _Share_ и любой неизвестный ключ).
|
||
|
||
Зеркалит CASE series_id-логику миграции 123_macro_indicator.sql (держать в синхроне).
|
||
"""
|
||
if not series_id:
|
||
return None
|
||
return _CBR_SERIES_MAP.get(series_id)
|
||
|
||
|
||
# ── Reader ──────────────────────────────────────────────────────────────────────
|
||
|
||
def get_macro_series(
|
||
db: Session,
|
||
indicator_type: str,
|
||
region: str = "rf",
|
||
since: date | None = None,
|
||
) -> list[tuple[date, float]]:
|
||
"""Весь ряд (obs_date, value) для (indicator_type, region), ASC по obs_date.
|
||
|
||
Args:
|
||
db: SQLAlchemy sync Session.
|
||
indicator_type: напр. 'mortgage_rate_weighted', 'key_rate' (key_rate заполнит PR B).
|
||
region: 'rf' | 'sverdl' | ... (по умолчанию 'rf').
|
||
since: если задано — только obs_date >= since.
|
||
|
||
Returns:
|
||
Список (obs_date, value) по возрастанию даты; строки с NULL value отброшены.
|
||
"""
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT obs_date, value
|
||
FROM macro_indicator
|
||
WHERE indicator_type = CAST(:itype AS text)
|
||
AND region = CAST(:region AS text)
|
||
AND value IS NOT NULL
|
||
AND (CAST(:since AS date) IS NULL OR obs_date >= CAST(:since AS date))
|
||
ORDER BY obs_date ASC
|
||
"""
|
||
),
|
||
{"itype": indicator_type, "region": region, "since": since},
|
||
).all()
|
||
return [(r[0], float(r[1])) for r in rows]
|
||
|
||
|
||
def get_latest_macro(
|
||
db: Session,
|
||
indicator_type: str,
|
||
region: str = "rf",
|
||
) -> tuple[date, float] | None:
|
||
"""Последняя (по obs_date) непустая точка ряда или None, если ряда нет.
|
||
|
||
Args:
|
||
db: SQLAlchemy sync Session.
|
||
indicator_type: тип показателя.
|
||
region: регион (по умолчанию 'rf').
|
||
|
||
Returns:
|
||
(obs_date, value) самой свежей точки, либо None.
|
||
"""
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT obs_date, value
|
||
FROM macro_indicator
|
||
WHERE indicator_type = CAST(:itype AS text)
|
||
AND region = CAST(:region AS text)
|
||
AND value IS NOT NULL
|
||
ORDER BY obs_date DESC
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"itype": indicator_type, "region": region},
|
||
).first()
|
||
if row is None:
|
||
return None
|
||
return row[0], float(row[1])
|