gendesign/backend/app/services/site_finder/macro.py

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


# ── Canonical region per indicator_type ───────────────────────────────────────────
# Реальная раскладка данных ЦБ в macro_indicator (проверено на prod):
#   • семейство CBR-ипотеки (формы 02, ИЖК) лежит под region='sverdl' — оно
#     бэкфиллится из cbr_mortgage_series, где единственный регион — 'sverdl'
#     (см. 123_macro_indicator.sql §2: region берётся из источника КАК ЕСТЬ);
#   • key_rate (и прочие национальные ряды) — под region='rf'.
# Это НЕ опечатка в данных, а фактический layout. Поэтому дефолтный регион ридера
# должен быть indicator-aware: иначе чтение mortgage_*-ряда без явного region молча
# вернёт пусто (footgun, #945 follow-up). Множество ипотечных типов берём как
# values() из _CBR_SERIES_MAP — единый источник правды, синхронный с миграцией.
_MORTGAGE_INDICATOR_TYPES: frozenset[str] = frozenset(
    itype for itype, _unit in _CBR_SERIES_MAP.values()
)

# Региональные ряды Росстата (#946) кладутся под region='sverdl' (субъект-агрегат) и
# 'ekb' (г. Екатеринбург), НЕ под 'rf'. Два источника:
#   • open-data (source='rosstat'): population_total (ОКТМО 65 / ЕКБ);
#   • ЕМИСС/fedstat (source='emiss'): income_per_capita (среднедушевые доходы, ОКАТО 65 —
#     только субъект, ЕКБ-среза в ОКАТО-размерности ЕМИСС нет).
# Дефолтный region таких типов = 'sverdl' (субъект — основной уровень потребления
# downstream affordability/forecast). Явный region='ekb' читатель задаёт сам.
# Источник правды по типам — DatasetSpec/EmissIndicatorSpec в rosstat_emiss (синхронить).
_ROSSTAT_SVERDL_INDICATOR_TYPES: frozenset[str] = frozenset(
    {"population_total", "income_per_capita"}
)


def _canonical_region(indicator_type: str) -> str:
    """Канонический region для indicator_type, когда вызыватель не задал явный.

    Детерминированно, без БД:
      • 6 CBR-ипотечных типов (mortgage_*) → 'sverdl' (так лежат на prod);
      • региональные ряды Росстата (population_total, income_per_capita, #946) → 'sverdl';
      • всё остальное (key_rate и пр.) → 'rf'.
    Источники истины — _CBR_SERIES_MAP (зеркало миграции 123) и _ROSSTAT_SVERDL_INDICATOR_TYPES
    (зеркало DatasetSpec'ов rosstat_emiss).
    """
    if indicator_type in _MORTGAGE_INDICATOR_TYPES:
        return "sverdl"
    if indicator_type in _ROSSTAT_SVERDL_INDICATOR_TYPES:
        return "sverdl"
    return "rf"


# ── Reader ──────────────────────────────────────────────────────────────────────


def get_macro_series(
    db: Session,
    indicator_type: str,
    region: str | None = None,
    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'.
        region: 'rf' | 'sverdl' | ... Если None — резолвится indicator-aware дефолт
            через _canonical_region (CBR-ипотека → 'sverdl', остальное → 'rf'),
            т.к. на prod эти семейства лежат в РАЗНЫХ регионах. Явный region
            ВСЕГДА выигрывает (вызыватель может переопределить дефолт).
        since: если задано — только obs_date >= since.

    Returns:
        Список (obs_date, value) по возрастанию даты; строки с NULL value отброшены.
    """
    if region is None:
        region = _canonical_region(indicator_type)
    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 | None = None,
) -> tuple[date, float] | None:
    """Последняя (по obs_date) непустая точка ряда или None, если ряда нет.

    Args:
        db: SQLAlchemy sync Session.
        indicator_type: тип показателя.
        region: 'rf' | 'sverdl' | ... Если None — резолвится indicator-aware дефолт
            через _canonical_region (CBR-ипотека → 'sverdl', остальное → 'rf'):
            на prod эти семейства лежат в РАЗНЫХ регионах. Явный region ВСЕГДА
            выигрывает (вызыватель может переопределить дефолт).

    Returns:
        (obs_date, value) самой свежей точки, либо None.
    """
    if region is None:
        region = _canonical_region(indicator_type)
    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])