gendesign/tradein-mvp/backend/app/services/cbr_macro.py

"""Макро-ряды ЦБ РФ: ипотека по субъектам (XLSX) + ключевая ставка (SOAP).

CONTEXT: продукту нужен региональный макро-контекст ипотечного рынка (динамика ставок,
объёмов выдач и задолженности по субъектам РФ) и дневная ключевая ставка. Этот модуль
только собирает данные в cbr_mortgage_series / cbr_key_rate (292_cbr_macro_series.sql) —
подключение к estimator вынесено в отдельное продуктовое решение (out of scope).

ИСТОЧНИК (ипотека): три XLSX-дашборда cbr.ru, wide-формат (территория × месяц),
строка 1 пустая, строка 2 — заголовок-описание, строка 3 — шапка периодов
(«Январь 2019», «Февраль 2019», …), строки 4+ — территории в колонке A
(РФ целиком → федеральные округа → субъекты). Unpivot в длинный ряд
(region, period_month, series, value).

ИСТОЧНИК (ключевая ставка): POST https://www.cbr.ru/DailyInfoWebServ/DailyInfo.asmx,
SOAP 1.1 (Content-Type: text/xml; charset=utf-8, SOAPAction: http://web.cbr.ru/KeyRate),
метод KeyRate(fromDate, ToDate). Ответ — DataSet-XML со строками
<KR><DT>дата</DT><Rate>значение</Rate></KR> (namespace-агностичный парсинг: берём по
локальному имени тега, т.к. .NET DataSet оборачивает их в diffgram/reset-default-ns).

TLS: cbr.ru отдаёт RU-сертификат → verify=False (open data, без auth/PII).
psycopg v3: CAST(:x AS type), НИКОГДА :x::type.

Дизайн-инвариант (как domrf_kapremont_loader): сервис-функции НЕ коммитят — коммитит
caller (app/tasks/cbr_macro_pull.py). Изоляция сбоев — try/except на серию,
db.rollback() в except (иначе аборченная транзакция каскадит на следующие серии).
"""

from __future__ import annotations

import io
import logging
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import date, datetime

import httpx
import openpyxl
from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

DOWNLOAD_TIMEOUT_SEC = 60.0

# ---------------------------------------------------------------------------
# Ипотека по субъектам — конфиг серий
# ---------------------------------------------------------------------------

_MORTGAGE_BASE_URL = "https://www.cbr.ru/vfs/statistics/BankSector/Mortgage"
NEW_LOANS_URL = f"{_MORTGAGE_BASE_URL}/02_11_New_loans_mortgage.xlsx"
RATES_URL = f"{_MORTGAGE_BASE_URL}/02_13_Rates_mortgage.xlsx"
DEBT_URL = f"{_MORTGAGE_BASE_URL}/02_14_Debt_mortgage.xlsx"


@dataclass(frozen=True)
class CbrMortgageSeries:
    slug: str
    url: str
    sheet: str


# По одной серии на файл (лист «в рублях» — основной ряд каждого дашборда).
CBR_MORTGAGE_SERIES: list[CbrMortgageSeries] = [
    CbrMortgageSeries("new_loans_rub", NEW_LOANS_URL, "в рублях"),
    CbrMortgageSeries("rate_rub", RATES_URL, "ставка в рублях"),
    CbrMortgageSeries("debt_rub", DEBT_URL, "в рублях"),
]


@dataclass(slots=True, frozen=True)
class CbrMortgageRow:
    region: str
    period_month: date
    series: str
    value: float


# ---------------------------------------------------------------------------
# Период: русские имена месяцев → date(y, m, 1)
# ---------------------------------------------------------------------------

RU_MONTHS: dict[str, int] = {
    "Январь": 1,
    "Февраль": 2,
    "Март": 3,
    "Апрель": 4,
    "Май": 5,
    "Июнь": 6,
    "Июль": 7,
    "Август": 8,
    "Сентябрь": 9,
    "Октябрь": 10,
    "Ноябрь": 11,
    "Декабрь": 12,
}


def parse_ru_period(period_str: str) -> date:
    """Разобрать шапку периода «Январь 2019» → date(2019, 1, 1)."""
    parts = str(period_str).strip().split()
    if len(parts) != 2:
        raise ValueError(f"cbr_macro: не удалось разобрать период {period_str!r}")
    month_name, year_str = parts
    month = RU_MONTHS.get(month_name)
    if month is None:
        raise ValueError(f"cbr_macro: неизвестное имя месяца {month_name!r} в {period_str!r}")
    try:
        year = int(year_str)
    except ValueError as exc:
        raise ValueError(f"cbr_macro: неверный год в периоде {period_str!r}") from exc
    return date(year, month, 1)


def parse_period_cell(cell: object) -> date:
    """Разобрать ячейку шапки периода в первое число месяца.

    ЦБ отдаёт шапку в двух формах, зависящих от файла:
      * текст «Январь 2019» — 02_11_New_loans_mortgage, 02_13_Rates_mortgage;
      * настоящий datetime/date — 02_14_Debt_mortgage.
    Обе приводятся к date(y, m, 1).
    """
    if isinstance(cell, datetime):
        return date(cell.year, cell.month, 1)
    if isinstance(cell, date):
        return date(cell.year, cell.month, 1)
    return parse_ru_period(str(cell))


# ---------------------------------------------------------------------------
# Unpivot листа (чистая функция — тестируется без сети/файла)
# ---------------------------------------------------------------------------

HEADER_SCAN_ROWS = 6
"""Сколько первых строк просматривать в поисках шапки периодов."""

MIN_HEADER_PERIODS = 6
"""Минимум распознанных периодов, чтобы считать строку шапкой (а не данными)."""


def _locate_header_row(rows: list[tuple[object, ...]]) -> tuple[int, dict[int, date]]:
    """Найти строку шапки периодов и колонки-периоды в ней.

    Раскладка у файлов ЦБ РАЗНАЯ и жёстко фиксировать индекс строки нельзя:
      * 02_11 / 02_13 — строка 1 пустая, строка 2 — описание, шапка в строке 3 (idx 2);
      * 02_14 (задолженность) — строка 1 описание, шапка уже в строке 2 (idx 1),
        и периоды там datetime, а не «Январь 2019».
    Ранее индекс был захардкожен на idx 2, из-за чего серия задолженности молча
    давала ноль строк. Поэтому шапку ищем: берём строку с наибольшим числом
    распознанных периодов среди первых HEADER_SCAN_ROWS.
    """
    best_idx = -1
    best_periods: dict[int, date] = {}
    for row_idx, row in enumerate(rows[:HEADER_SCAN_ROWS]):
        periods: dict[int, date] = {}
        for col_idx, cell in enumerate(row):
            if col_idx == 0 or cell is None or str(cell).strip() == "":
                continue
            try:
                periods[col_idx] = parse_period_cell(cell)
            except ValueError:
                continue
        if len(periods) > len(best_periods):
            best_idx, best_periods = row_idx, periods
    if len(best_periods) < MIN_HEADER_PERIODS:
        return -1, {}
    return best_idx, best_periods


def parse_mortgage_sheet(
    rows: list[tuple[object, ...]], *, series_slug: str
) -> list[CbrMortgageRow]:
    """Unpivot wide-листа ЦБ (территория × месяц) в длинный ряд.

    rows — результат ws.iter_rows(values_only=True). Строка шапки периодов ищется
    динамически (`_locate_header_row`), территории идут сразу после неё.
    """
    if len(rows) < 3:
        return []

    header_idx, periods = _locate_header_row(rows)
    if header_idx < 0:
        logger.warning(
            "cbr_macro: шапка периодов не найдена для series=%s — лист пропущен", series_slug
        )
        return []

    out: list[CbrMortgageRow] = []
    for data_row in rows[header_idx + 1 :]:
        if not data_row or data_row[0] is None:
            continue
        region = str(data_row[0]).strip()
        if not region:
            continue
        for col_idx, period_month in periods.items():
            if col_idx >= len(data_row):
                continue
            raw_value = data_row[col_idx]
            if raw_value is None:
                continue
            try:
                value = float(raw_value)
            except (TypeError, ValueError):
                logger.warning(
                    "cbr_macro: пропуск ячейки region=%r period=%s series=%s — не число: %r",
                    region,
                    period_month,
                    series_slug,
                    raw_value,
                )
                continue
            out.append(
                CbrMortgageRow(
                    region=region, period_month=period_month, series=series_slug, value=value
                )
            )
    return out


# ---------------------------------------------------------------------------
# Загрузка XLSX + upsert
# ---------------------------------------------------------------------------

_UPSERT_MORTGAGE_SQL = text("""
    INSERT INTO cbr_mortgage_series (region, period_month, series, value, source, fetched_at)
    VALUES (
        CAST(:region AS text),
        CAST(:period_month AS date),
        CAST(:series AS text),
        CAST(:value AS double precision),
        'cbr',
        now()
    )
    ON CONFLICT (region, period_month, series)
    DO UPDATE SET
        value = EXCLUDED.value
        -- fetched_at НЕ трогаем (#2846 у sber_price_index): означает
        -- «когда мы ВПЕРВЫЕ увидели этот период», а не время последней загрузки.
""")


def _upsert_mortgage_rows(db: Session, rows: list[CbrMortgageRow]) -> int:
    for row in rows:
        db.execute(
            _UPSERT_MORTGAGE_SQL,
            {
                "region": row.region,
                "period_month": row.period_month.isoformat(),
                "series": row.series,
                "value": row.value,
            },
        )
    return len(rows)


def _download(url: str, *, client: httpx.Client) -> bytes:
    resp = client.get(url, timeout=DOWNLOAD_TIMEOUT_SEC)
    resp.raise_for_status()
    return resp.content


def load_cbr_mortgage_series(
    db: Session,
    series: CbrMortgageSeries,
    *,
    client: httpx.Client | None = None,
    dry_run: bool = False,
) -> dict[str, int]:
    """Скачать один XLSX-дашборд, разобрать один лист, upsert-нуть ряд.

    Не коммитит — коммитит caller.
    """
    own_client = client is None
    if own_client:
        # cbr.ru отдаёт RU-сертификат НУЦ Минцифры → verify=False. Открытые данные,
        # без auth/PII (прецедент: sber_index.py, domrf_kapremont_loader.py).
        client = httpx.Client(timeout=DOWNLOAD_TIMEOUT_SEC, verify=False)
    try:
        data = _download(series.url, client=client)
    finally:
        if own_client:
            client.close()

    wb = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True)
    try:
        if series.sheet not in wb.sheetnames:
            raise ValueError(
                f"cbr_macro: лист {series.sheet!r} отсутствует в {series.url} "
                f"(есть: {wb.sheetnames!r})"
            )
        ws = wb[series.sheet]
        rows_raw = list(ws.iter_rows(values_only=True))
    finally:
        wb.close()

    parsed = parse_mortgage_sheet(rows_raw, series_slug=series.slug)
    upserted = 0 if dry_run else _upsert_mortgage_rows(db, parsed)
    return {"rows": len(parsed), "upserted": upserted}


def pull_cbr_mortgage(
    db: Session,
    *,
    series_list: list[CbrMortgageSeries] | None = None,
    client: httpx.Client | None = None,
    dry_run: bool = False,
) -> dict[str, int]:
    """Забрать все (или выбранные) серии ипотечной статистики.

    Per-series try/except: одна сбойнувшая серия логируется и не роняет остальные
    (урок #1345 у sber_index: без rollback аборченная транзакция каскадит дальше).
    """
    if series_list is None:
        series_list = CBR_MORTGAGE_SERIES

    counters = {"upserted": 0, "skipped": 0, "errors": 0}
    own_client = client is None
    if own_client:
        client = httpx.Client(timeout=DOWNLOAD_TIMEOUT_SEC, verify=False)
    try:
        for series in series_list:
            try:
                result = load_cbr_mortgage_series(db, series, client=client, dry_run=dry_run)
                if result["rows"] == 0:
                    logger.info("cbr_macro: пустой результат для series=%s", series.slug)
                    counters["skipped"] += 1
                else:
                    counters["upserted"] += result["upserted"]
                    logger.info(
                        "cbr_macro: upserted %d rows for series=%s", result["upserted"], series.slug
                    )
            except Exception:
                db.rollback()
                logger.exception("cbr_macro: series=%s сбойнула — пропуск", series.slug)
                counters["errors"] += 1
    finally:
        if own_client:
            client.close()
    return counters


# ---------------------------------------------------------------------------
# Ключевая ставка — SOAP
# ---------------------------------------------------------------------------

CBR_SOAP_URL = "https://www.cbr.ru/DailyInfoWebServ/DailyInfo.asmx"
KEY_RATE_START = date(2013, 9, 13)  # инструмент введён Советом директоров ЦБ РФ

_KEY_RATE_ENVELOPE = """<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" \
xmlns:xsd="http://www.w3.org/2001/XMLSchema" \
xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
  <soap:Body>
    <KeyRate xmlns="http://web.cbr.ru/">
      <fromDate>{from_date}</fromDate>
      <ToDate>{to_date}</ToDate>
    </KeyRate>
  </soap:Body>
</soap:Envelope>"""


def build_key_rate_envelope(from_date: date, to_date: date) -> str:
    return _KEY_RATE_ENVELOPE.format(from_date=from_date.isoformat(), to_date=to_date.isoformat())


def _local_tag(tag: str) -> str:
    """Локальное имя тега без namespace-префикса ('{ns}KR' → 'KR')."""
    return tag.rsplit("}", 1)[-1]


def parse_key_rate_response(xml_text: str) -> list[tuple[date, float]]:
    """Разобрать SOAP-ответ KeyRate: строки <KR><DT>…</DT><Rate>…</Rate></KR>.

    Namespace-агностично (ищем по локальному имени тега) — .NET DataSet
    оборачивает строки в diffgram/reset-default-ns, точный путь не гарантирован.
    """
    root = ET.fromstring(xml_text)
    out: list[tuple[date, float]] = []
    for el in root.iter():
        if _local_tag(el.tag) != "KR":
            continue
        dt_text: str | None = None
        rate_text: str | None = None
        for child in el:
            name = _local_tag(child.tag)
            if name == "DT":
                dt_text = child.text
            elif name == "Rate":
                rate_text = child.text
        if not dt_text or rate_text is None:
            continue
        try:
            rate_date = datetime.fromisoformat(dt_text).date()
            rate = float(rate_text)
        except (ValueError, TypeError):
            logger.warning(
                "cbr_macro: пропуск строки KeyRate — не распознана: DT=%r Rate=%r",
                dt_text,
                rate_text,
            )
            continue
        out.append((rate_date, rate))
    return out


def fetch_key_rate(
    client: httpx.Client, *, from_date: date, to_date: date
) -> list[tuple[date, float]]:
    body = build_key_rate_envelope(from_date, to_date)
    headers = {
        "Content-Type": "text/xml; charset=utf-8",
        "SOAPAction": "http://web.cbr.ru/KeyRate",
    }
    resp = client.post(CBR_SOAP_URL, content=body.encode("utf-8"), headers=headers)
    resp.raise_for_status()
    return parse_key_rate_response(resp.text)


_UPSERT_KEY_RATE_SQL = text("""
    INSERT INTO cbr_key_rate (rate_date, rate, fetched_at)
    VALUES (CAST(:rate_date AS date), CAST(:rate AS double precision), now())
    ON CONFLICT (rate_date)
    DO UPDATE SET
        rate = EXCLUDED.rate
        -- fetched_at НЕ трогаем — та же логика, что и у cbr_mortgage_series.
""")


def _upsert_key_rate_rows(db: Session, rows: list[tuple[date, float]]) -> int:
    for rate_date, rate in rows:
        db.execute(_UPSERT_KEY_RATE_SQL, {"rate_date": rate_date.isoformat(), "rate": rate})
    return len(rows)


def load_key_rate(
    db: Session,
    *,
    from_date: date | None = None,
    to_date: date | None = None,
    client: httpx.Client | None = None,
    dry_run: bool = False,
) -> dict[str, int]:
    """Скачать ключевую ставку за диапазон дат, upsert-нуть в cbr_key_rate.

    Не коммитит — коммитит caller. from_date/to_date по умолчанию — вся история
    инструмента (KEY_RATE_START) до сегодня.
    """
    if from_date is None:
        from_date = KEY_RATE_START
    if to_date is None:
        to_date = date.today()

    own_client = client is None
    if own_client:
        client = httpx.Client(timeout=DOWNLOAD_TIMEOUT_SEC, verify=False)
    try:
        rows = fetch_key_rate(client, from_date=from_date, to_date=to_date)
    finally:
        if own_client:
            client.close()

    upserted = 0 if dry_run else _upsert_key_rate_rows(db, rows)
    result = {"rows": len(rows), "upserted": upserted}
    logger.info("cbr_macro: key_rate load DONE (dry_run=%s): %s", dry_run, result)
    return result