Light1YT
ea42d29a4a
Site Finder v2 / GG-форсайт. Deterministic, no LLM. Fills the key_rate series (empty after PR A) into macro_indicator from the live CBR source. - services/scrapers/cbr_macro.py: fetch_key_rate via CBR SOAP DailyInfoWebServ KeyRateXML(fromDate, ToDate) (POST soap+xml; ToDate capital-T + dateTime suffix required — flat GET returns empty). Pure parse_key_rate_xml (TZ-safe date, empty/malformed → []). Live-verified: 85 rows, current rate 14.5%. - workers/tasks/cbr_macro_sync.py: upsert (key_rate, rf, daily, %) into macro_indicator ON CONFLICT DO UPDATE; SAVEPOINT per row; fetch-fail → raise. - beat: cbr-macro-sync-weekly (Mon 05:30 МСК, Europe/Moscow tz). - 15 tests (pure parse + mocked upsert). No migration (table from #963 PR A).
115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
"""Celery task: синхронизация макро-показателей ЦБ (ключевая ставка) в
|
||
``macro_indicator`` (#945 PR B, GG-форсайт / Site Finder v2).
|
||
|
||
Тянет историю ключевой ставки через ``fetch_key_rate`` (живой SOAP CBR
|
||
DailyInfoWebServ) и апсертит каждую (дату, ставку) в ``macro_indicator`` как
|
||
ряд ``indicator_type='key_rate'``, ``region='rf'``, ``source='cbr'``.
|
||
|
||
Детерминированно, без LLM. ON CONFLICT делает повторные прогоны идемпотентными
|
||
(re-run обновляет value + updated_at). Расписание — еженедельно (ключевая
|
||
ставка меняется редко), регистрируется в beat_schedule.py.
|
||
|
||
Mirror conventions: ``SessionLocal()`` + try/finally close, ``logger`` (не
|
||
print), SAVEPOINT per-row в цикле upsert (backend.md), ``CAST(:x AS type)`` —
|
||
никогда ``:x::type`` (psycopg v3). На fetch-фейле — логируем и пробрасываем
|
||
(surfaces в Celery/GlitchTip), не глотаем.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.cbr_macro import fetch_key_rate
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# psycopg v3: CAST(:x AS type) — НИКОГДА :x::type (SQLAlchemy+psycopg3 роняет
|
||
# синтаксис на ::). Контракт колонок совпадает с macro_indicator (migration 123):
|
||
# (indicator_type, region, obs_date, value, source, frequency, unit, comment,
|
||
# updated_at) PK (indicator_type, region, obs_date).
|
||
UPSERT_KEY_RATE_SQL = text(
|
||
"""
|
||
INSERT INTO macro_indicator (
|
||
indicator_type, region, obs_date, value,
|
||
source, frequency, unit, comment
|
||
) VALUES (
|
||
'key_rate', 'rf', CAST(:d AS date), CAST(:v AS numeric),
|
||
'cbr', 'daily', '%', 'CBR key rate'
|
||
)
|
||
ON CONFLICT (indicator_type, region, obs_date) DO UPDATE SET
|
||
value = EXCLUDED.value,
|
||
updated_at = now()
|
||
"""
|
||
)
|
||
|
||
|
||
def _upsert_key_rate(db: Session, rows: list[tuple[date, Decimal]]) -> int:
|
||
"""Апсертит (date, rate) в macro_indicator. SAVEPOINT per-row, чтобы один
|
||
битый ряд не откатывал всю транзакцию. Возвращает число успешных upsert'ов."""
|
||
upserted = 0
|
||
for d, v in rows:
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(UPSERT_KEY_RATE_SQL, {"d": d, "v": v})
|
||
upserted += 1
|
||
except Exception as e:
|
||
logger.warning("upsert key_rate %s=%s failed: %s", d, v, e)
|
||
db.commit()
|
||
return upserted
|
||
|
||
|
||
@celery_app.task(
|
||
bind=True,
|
||
name="tasks.cbr_macro_sync.cbr_macro_sync",
|
||
max_retries=2,
|
||
)
|
||
def cbr_macro_sync(
|
||
self: Any,
|
||
from_date: str | None = None,
|
||
to_date: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Загрузить историю ключевой ставки ЦБ и апсертить в macro_indicator.
|
||
|
||
Args:
|
||
from_date: ISO-дата начала периода ('YYYY-MM-DD'). None → дефолт
|
||
scraper'а (2019-01-01, полный бэкфилл).
|
||
to_date: ISO-дата конца периода. None → сегодня.
|
||
|
||
Returns:
|
||
Счётчики ``{"fetched": N, "upserted": M, "from_date": ..., "to_date": ...}``.
|
||
|
||
Raises:
|
||
Пробрасывает любую ошибку fetch'а (CBRScraperError и пр.) — чтобы фейл
|
||
был виден в Celery / GlitchTip, а не молча проглочен.
|
||
"""
|
||
fd = date.fromisoformat(from_date) if from_date else None
|
||
td = date.fromisoformat(to_date) if to_date else None
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
rows = fetch_key_rate(from_date=fd, to_date=td)
|
||
fetched = len(rows)
|
||
logger.info("cbr_macro_sync: fetched %d key_rate rows", fetched)
|
||
|
||
upserted = _upsert_key_rate(db, rows)
|
||
logger.info("cbr_macro_sync: upserted %d/%d key_rate rows", upserted, fetched)
|
||
|
||
return {
|
||
"fetched": fetched,
|
||
"upserted": upserted,
|
||
"from_date": fd.isoformat() if fd else None,
|
||
"to_date": td.isoformat() if td else None,
|
||
}
|
||
except Exception as e:
|
||
logger.exception("cbr_macro_sync failed: %s", e)
|
||
raise
|
||
finally:
|
||
db.close()
|