The user wants a data analytics project in Spanish covering Argentine government debt using public BCRA and National Treasury data, delivered as PDF and HTML. Since I can't fetch live data directly, I'm planning to build a complete working scaffold that pulls from official public APIs like the BCRA API v3. I'm designing the project structure now: extraction modules for BCRA, finance ministry, and datos.gob.ar sources, transform/clean/metrics layers, a warehouse loader, visualization charts, and HTML/PDF report builders, plus notebooks, data folders, a Makefile, and Docker setup to make it fully reproducible. I'm cataloging the real BCRA API endpoints and key monetary variables -- international reserves, exchange rates, LELIQ/pases rates, monetary base, and deposits -- so the extraction scripts pull actual series rather than placeholder data. I'm also noting the quarterly debt presentation and "Datos de deuda" datasets from Economía, plus the Informe Mensual de la Deuda and the datos.gob.ar time series API with useful IDs like exchange rate and public debt series. Rather than hardcoding possibly incorrect series IDs, I should use the search endpoint so the code discovers real IDs dynamically — that's the more honest and robust approach. I'll also include Oficina de Presupuesto del Congreso debt datasets and the Ministerio de Economía monthly debt report, building the code around discovery endpoints instead of fragile hardcoded IDs. Now I'm mapping out the key analytics for debt analysis: gross central administration debt (USD and % of GDP), currency composition, legal jurisdiction composition, instrument type breakdown, interest rate type composition, maturity profile by year, average life/duration, and year-over-year and monthly variation decomposed into new issuance versus other factors. I'm also factoring in intra-public-sector debt versus market debt, BCRA's remunerated liabilities (LELIQ/Pases/LEFI) as quasi-fiscal debt, consolidated debt (Treasury + BCRA), debt-to-GDP and debt-to-exports ratios, and interest-to-revenue ratios, along with a decomposition identity breaking down debt change into net issuance, FX valuation effects, CER indexation effects, capitalized interest, and other factors. Now I'm planning the full project structure: a README covering architecture and sources, requirements.txt, a sources config file, ETL modules, a metrics module, a self-contained HTML report template with embedded charts, a PDF builder using WeasyPrint with a Playwright fallback, a Makefile, a data dictionary, quality checks, and methodology notes. For visualization, I'm leaning toward generating charts with Plotly for HTML interactivity and rendering static versions for the PDF output. I'm thinking through the BCRA API v3 endpoint structure, mapping out the fields it returns like variable IDs and descriptions for monetary statistics, so the ETL code pulls the right data correctly. I want to keep the full answer well-organized and complete, writing it directly with code blocks rather than relying on separate artifacts, and keeping each code snippet tight but functional. I'm noting that BCRA's TLS certificate can be flaky, so I'll make SSL verification configurable rather than hardcoding it off, and I should keep requests gentle to respect rate limits. I'm also starting to map out the datos.gob.ar series API structure as a complementary data source. For the Secretaría de Finanzas debt data, since the file URLs shift over time, I'll build a small scraper that crawls the landing page for Excel or CSV links rather than hardcoding paths. I'm also planning to layer in DuckDB as a lightweight SQL warehouse to make the analytics side feel more credible, and now I'm starting to outline the final answer structure beginning with what the project delivers. Now I'm listing out the remaining sections: data sources table, architecture, folder structure, file-by-file code, data model and dictionary, metrics methodology, execution instructions, quality validations, caveats, and roadmap. Writing requirements.txt... Next I'm sketching the HTTP module with retry/caching logic for raw data, then drafting the BCRA extraction module with functions for fetching the catalog and time series data from the API. I'm thinking through pagination handling for the v3 API's results/metadata structure, plus adding FX rate extraction from the Cotizaciones endpoint. For balance sheet variables like LELIQ and pases pasivos, I'll avoid hardcoding IDs and instead match against the catalog by regex on the description field, which is more robust against ID changes. I'm drafting the config file with regex patterns to match each variable name against the BCRA catalog descriptions—reserves, monetary base, exchange rates, CER, UVA, policy rate, repos, and total deposits. Now I'm working on the code that resolves these regex matches against the catalog to get variable IDs, logging matches and writing a resolution table into the report for transparency. Then I'm building a scraper in `finanzas.py` that crawls specific government debt pages looking for spreadsheet and zip file links to download into a raw data folder, followed by a transform script to normalize the Treasury debt data. Since sheet layouts vary unpredictably, I want this normalizer to support configurable column mapping plus an inspect mode that dumps sheet names and headers so users can map them manually in YAML — that feels more honest than guessing. I'm also planning to pull in datos.gob.ar series for GDP, exchange rate, inflation, tax collection, and fiscal results, using a search endpoint for discovery while allowing specific series IDs to be pinned in YAML configuration, and jotting down a few known-good series IDs I'm fairly confident about. Given my uncertainty about some of these IDs, I'll treat them as candidates and have the code verify them against the API, falling back to search if invalid, with a clear README note telling users to confirm IDs themselves — a safer engineering approach than asserting facts I'm not sure of. Now I'm moving on to the metrics module and its decomposition. I'm designing the Jinja report template with KPI cards, sections, charts, methodology, sources, lineage, and a changelog, styled for PDF via @page CSS. Since WeasyPrint can't render interactive Plotly JS, I'll have the template support both HTML and PDF modes, swapping in static Kaleido-rendered images for PDF output. I'm also planning the Makefile targets, a pandera-based quality checks module with results shown in the report, and a GitHub Actions workflow for monthly refreshes. Now I'm drafting the methodology section covering debt variation decomposition — nominal residual times price/adjustment times exchange rate, broken into net issuance, FX effect, indexation effect, capitalized interest, and residual components. I'm also noting the consolidated SPN debt caveats, particularly the methodological complications around BCRA remunerated liabilities after the 2024 LEFI changes, before moving into the debt ratio definitions. Beyond debt/GDP, I'm specifying interest burden, foreign-currency debt over reserves, average life calculations, and maturity profiles. I'll also point to OPC's open data portal and INDEC as sources for GDP and external debt figures, and plan to include a methodology doc alongside the main deliverable. I should note licensing—datos.gob.ar generally follows Open Data terms under Ley 27.275, while BCRA has its own usage terms—so I'll mention attribution requirements generally rather than citing exact license text. # 📊 Proyecto de Data Analytics — Deuda Pública Argentina **`deuda-ar-analytics`** · Pipeline ETL + warehouse analítico + informe HTML interactivo y PDF imprimible, construido 100 % sobre datos públicos del **BCRA**, la **Secretaría de Finanzas / Tesoro Nacional**, **datos.gob.ar**, **INDEC** y la **OPC**. --- ## 1. Qué entrega el proyecto | Entregable | Formato | Contenido | |---|---|---| | `outputs/html/informe_deuda.html` | HTML autocontenido (Plotly) | Informe interactivo con 6 secciones, KPIs, 14 gráficos, tablas filtrables | | `outputs/pdf/informe_deuda.pdf` | PDF A4 paginado | Mismo informe, gráficos vectoriales SVG, portada, índice, anexo metodológico | | `data/processed/deuda.duckdb` | DuckDB | Warehouse analítico consultable con SQL | | `data/processed/*.parquet` | Parquet | Tablas tidy listas para BI (Power BI / Metabase / Superset) | | `outputs/quality/report.html` | HTML | Reporte de validaciones de calidad de datos | | `docs/metodologia.md` | Markdown | Fórmulas, supuestos, limitaciones | --- ## 2. Fuentes de datos públicas > ⚠️ **Nota importante:** las URLs de los archivos Excel de Finanzas y los IDs de series de `datos.gob.ar` **cambian con el tiempo**. Por eso el proyecto **no hardcodea** nada frágil: incluye un comando `make discover` que descubre endpoints e IDs vivos y los escribe en `config/resolved.yml`. Verificá siempre la salida de ese paso antes de publicar un informe. ### 2.1 BCRA — API oficial | Recurso | Endpoint | Frecuencia | |---|---|---| | Catálogo de variables monetarias | `GET https://api.bcra.gob.ar/estadisticas/v3.0/monetarias` | — | | Serie histórica por variable | `GET .../v3.0/monetarias/{idVariable}?desde=&hasta=&limit=&offset=` | Diaria/mensual | | Cotizaciones | `GET https://api.bcra.gob.ar/estadisticascambiarias/v1.0/Cotizaciones` | Diaria | | Deuda externa privada | `GET https://api.bcra.gob.ar/estadisticas/v1.0/...` (ver catálogo) | Trimestral | Variables clave que usamos (se resuelven **por regex sobre `descripcion`**, no por ID fijo): Reservas internacionales · Base monetaria total · Pasivos remunerados (LELIQ/Pases/LEFI) · Tipo de cambio mayorista A3500 · CER · UVA · Tasa de política monetaria · Depósitos del sector público. Complemento no-API (descarga de planillas): *Boletín Estadístico*, *Base monetaria y factores de explicación*, *Balance semanal resumido* en `bcra.gob.ar/PublicacionesEstadisticas/`. ### 2.2 Secretaría de Finanzas — Tesoro Nacional | Recurso | Página | Formato | |---|---|---| | **Datos de deuda** (planilla maestra por instrumento) | `argentina.gob.ar/economia/finanzas/datos-deuda` | XLSX | | **Presentación gráfica de la deuda** (trimestral) | `.../presentacion-grafica-de-deuda` | XLSX + PDF | | **Informe mensual de deuda (IMD)** | `.../informes-deuda` | XLSX + PDF | | Resultados de licitaciones de Letras y Bonos | `.../licitaciones-de-deuda` | XLSX | | Perfil de vencimientos de capital e intereses | dentro de la presentación gráfica | XLSX | ### 2.3 Otras fuentes complementarias | Fuente | Uso | Endpoint | |---|---|---| | **datos.gob.ar — Series de Tiempo AR** | PIB, IPC, TC, recaudación, resultado fiscal | `https://apis.datos.gob.ar/series/api/series/?ids=...` · búsqueda: `/search/?q=deuda` | | **INDEC** | PIB corriente, Posición de Inversión Internacional, deuda externa bruta total | `indec.gob.ar` (XLS) | | **OPC (Oficina de Presupuesto del Congreso)** | Datasets de deuda, servicios proyectados, informes de validación cruzada | `opc.gob.ar` → Datos abiertos | | **Ministerio de Economía — Sector Público** | Resultado fiscal base caja, intereses pagados | `argentina.gob.ar/economia/sechacienda/...` | **Atribución:** todos los datasets son de acceso público bajo el marco de la Ley 27.275 y el Decreto 117/2016 (datos abiertos). El informe incluye una sección de fuentes con URL + fecha de acceso + hash SHA-256 de cada archivo descargado. --- ## 3. Arquitectura ``` ┌──────────────────────────────────────────┐ │ EXTRACT (src/extract) │ BCRA API v3 ─────▶│ bcra.py → data/raw/bcra/ │ Finanzas XLSX ───▶│ finanzas.py → data/raw/finanzas/ │ datos.gob.ar ────▶│ datos_gob.py → data/raw/datosgob/ │ INDEC / OPC ─────▶│ indec.py / opc.py │ └───────────────┬──────────────────────────┘ │ (cache + manifest.json con SHA256) ┌───────────────▼──────────────────────────┐ │ TRANSFORM (src/transform) │ │ clean.py → tidy long tables │ │ metrics.py→ descomposición, ratios, │ │ perfil de vencimientos │ └───────────────┬──────────────────────────┘ │ Parquet ┌───────────────▼──────────────────────────┐ │ LOAD (src/load) → DuckDB │ │ stg_* → fct_* → mart_* (SQL en sql/) │ └───────────────┬──────────────────────────┘ │ ┌───────────────────────┼───────────────────────┐ ▼ ▼ ▼ QUALITY (pandera) VIZ (Plotly) REPORT (Jinja2) outputs/quality/ figs HTML + SVG ├─ HTML (plotly.js CDN) └─ PDF (WeasyPrint) ``` --- ## 4. Estructura de carpetas ``` deuda-ar-analytics/ ├── README.md ├── requirements.txt ├── Makefile ├── config/ │ ├── fuentes.yml # configuración declarativa de fuentes │ ├── mapeos.yml # mapeo de columnas de planillas Finanzas │ └── resolved.yml # (generado) IDs/URLs descubiertos ├── sql/ │ ├── 01_staging.sql │ ├── 02_facts.sql │ └── 03_marts.sql ├── src/ │ ├── common/ │ │ ├── http.py # sesión con reintentos, cache, manifest │ │ ├── logging_cfg.py │ │ └── utils.py │ ├── extract/ │ │ ├── bcra.py │ │ ├── finanzas.py │ │ ├── datos_gob.py │ │ └── discover.py │ ├── transform/ │ │ ├── clean.py │ │ └── metrics.py │ ├── load/ │ │ └── warehouse.py │ ├── quality/ │ │ └── checks.py │ ├── viz/ │ │ └── charts.py │ ├── report/ │ │ ├── templates/ │ │ │ ├── informe.html.j2 │ │ │ └── styles.css │ │ ├── build_html.py │ │ └── build_pdf.py │ └── cli.py ├── notebooks/01_exploracion.ipynb ├── docs/metodologia.md ├── data/{raw,interim,processed}/ ├── outputs/{html,pdf,figs,quality}/ └── .github/workflows/refresh.yml ``` --- ## 5. Código ### `requirements.txt` ```txt pandas>=2.2 numpy>=1.26 requests>=2.32 urllib3>=2.2 PyYAML>=6.0 openpyxl>=3.1 xlrd>=2.0.1 duckdb>=1.0 pandera>=0.20 plotly>=5.22 kaleido==0.2.1 Jinja2>=3.1 weasyprint>=62 beautifulsoup4>=4.12 lxml>=5.2 python-dateutil>=2.9 tqdm>=4.66 typer>=0.12 rich>=13.7 ``` --- ### `config/fuentes.yml` ```yaml # ============================================================ # Configuración declarativa de fuentes de datos públicas # ============================================================ general: fecha_inicio: "2015-01-01" moneda_base: "USD" tz: "America/Argentina/Buenos_Aires" user_agent: "deuda-ar-analytics/1.0 (proyecto educativo de data analytics)" # El certificado TLS del BCRA puede fallar en algunos entornos. # Poner en false SOLO si es necesario y documentarlo. verify_ssl_bcra: true request_timeout: 60 cache_ttl_hours: 24 bcra: base_url: "https://api.bcra.gob.ar" catalogo: "/estadisticas/v3.0/monetarias" serie: "/estadisticas/v3.0/monetarias/{id}" cotizaciones: "/estadisticascambiarias/v1.0/Cotizaciones" page_limit: 3000 # Resolución por REGEX sobre el campo `descripcion` del catálogo. # Más robusto que hardcodear idVariable (que el BCRA puede reordenar). variables: reservas_internacionales: "(?i)reservas\\s+internacionales" base_monetaria: "(?i)base\\s+monetaria.*total" circulacion_monetaria: "(?i)circulaci[oó]n\\s+monetaria" pases_pasivos: "(?i)pases\\s+pasivos" leliq: "(?i)leliq|letras\\s+de\\s+liquidez" tc_mayorista: "(?i)tipo\\s+de\\s+cambio\\s+mayorista" tc_minorista: "(?i)tipo\\s+de\\s+cambio\\s+minorista" cer: "(?i)^cer\\b|coeficiente\\s+de\\s+estabilizaci[oó]n" uva: "(?i)unidad\\s+de\\s+valor\\s+adquisitivo|^uva" tasa_politica_monetaria: "(?i)tasa\\s+de\\s+pol[ií]tica\\s+monetaria" badlar: "(?i)badlar.*bancos\\s+privados" depositos_sector_publico: "(?i)dep[oó]sitos.*sector\\s+p[uú]blico" finanzas: # Páginas que se scrapean en busca de XLSX/CSV publicados. paginas: - "https://www.argentina.gob.ar/economia/finanzas/datos-deuda" - "https://www.argentina.gob.ar/economia/finanzas/presentacion-grafica-de-deuda" - "https://www.argentina.gob.ar/economia/finanzas/informes-deuda" - "https://www.argentina.gob.ar/economia/finanzas/licitaciones-de-deuda" extensiones: [".xlsx", ".xls", ".csv", ".zip"] # Palabras clave para clasificar los archivos descargados clasificador: deuda_instrumentos: "(?i)instrumento|detalle.*deuda|stock" presentacion_grafica: "(?i)presentaci[oó]n.*gr[aá]fic" vencimientos: "(?i)vencimiento|perfil|servicio" licitaciones: "(?i)licitaci|resultado.*letra" datos_gob: base_url: "https://apis.datos.gob.ar/series/api" # IDs "candidatos": el pipeline los VERIFICA contra la API. # Si alguno falla, cae al endpoint /search y lo reemplaza. series: tc_nominal: candidatos: ["168.1_T_CAMBIOR_D_0_0_26"] busqueda: "tipo de cambio nominal promedio mensual" pib_corriente: candidatos: [] busqueda: "producto interno bruto precios corrientes trimestral" ipc_nivel_general: candidatos: [] busqueda: "indice de precios al consumidor nivel general nacional" recaudacion_total: candidatos: [] busqueda: "recaudacion tributaria total mensual" resultado_primario: candidatos: [] busqueda: "resultado primario sector publico nacional no financiero" analisis: # Umbrales para semáforos y alertas en el informe umbrales: deuda_pib_alerta: 80.0 # % del PIB fx_share_alerta: 65.0 # % de deuda en moneda extranjera vida_promedio_min: 4.0 # años cobertura_reservas_min: 1.0 # reservas / vencimientos FX 12m ``` --- ### `src/common/http.py` ```python """Sesión HTTP con reintentos, cache en disco y manifest de trazabilidad.""" from __future__ import annotations import hashlib import json import logging import time from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any import requests import urllib3 from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry log = logging.getLogger(__name__) MANIFEST_PATH = Path("data/raw/_manifest.json") def build_session(user_agent: str, verify: bool = True) -> requests.Session: s = requests.Session() retry = Retry( total=5, backoff_factor=1.5, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(["GET", "HEAD"]), respect_retry_after_header=True, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10) s.mount("https://", adapter) s.mount("http://", adapter) s.headers.update({"User-Agent": user_agent, "Accept": "application/json, */*"}) s.verify = verify if not verify: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) log.warning("TLS verification DESACTIVADA para esta sesión. Documentar en el informe.") return s # ----------------------------------------------------------------- manifest def _load_manifest() -> dict[str, Any]: if MANIFEST_PATH.exists(): return json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) return {} def _save_manifest(m: dict[str, Any]) -> None: MANIFEST_PATH.parent.mkdir(parents=True, exist_ok=True) MANIFEST_PATH.write_text(json.dumps(m, indent=2, ensure_ascii=False), encoding="utf-8") def register(url: str, dest: Path, status: int, note: str = "") -> dict[str, Any]: """Registra la descarga con hash SHA-256 → reproducibilidad y citabilidad.""" m = _load_manifest() sha = hashlib.sha256(dest.read_bytes()).hexdigest() if dest.exists() else None entry = { "url": url, "archivo": str(dest), "descargado_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"), "status": status, "bytes": dest.stat().st_size if dest.exists() else 0, "sha256": sha, "nota": note, } m[str(dest)] = entry _save_manifest(m) return entry def manifest_df(): import pandas as pd return pd.DataFrame(_load_manifest().values()) # ----------------------------------------------------------------- fetchers def get_json(session: requests.Session, url: str, params: dict | None = None, timeout: int = 60, pause: float = 0.35) -> dict: log.debug("GET %s %s", url, params or "") r = session.get(url, params=params, timeout=timeout) r.raise_for_status() time.sleep(pause) # cortesía con la API pública return r.json() def download(session: requests.Session, url: str, dest: Path, ttl_hours: int = 24, timeout: int = 120) -> Path: """Descarga con cache por TTL. Devuelve la ruta local.""" dest.parent.mkdir(parents=True, exist_ok=True) if dest.exists(): age = datetime.now() - datetime.fromtimestamp(dest.stat().st_mtime) if age < timedelta(hours=ttl_hours): log.info("cache HIT %s (%.1f h)", dest.name, age.total_seconds() / 3600) return dest log.info("descargando %s", url) with session.get(url, stream=True, timeout=timeout) as r: r.raise_for_status() tmp = dest.with_suffix(dest.suffix + ".part") with tmp.open("wb") as fh: for chunk in r.iter_content(1 << 16): fh.write(chunk) tmp.replace(dest) register(url, dest, r.status_code) return dest ``` --- ### `src/extract/bcra.py` ```python """Extracción desde la API pública de Estadísticas del BCRA (v3.0).""" from __future__ import annotations import logging import re from pathlib import Path import pandas as pd from ..common.http import get_json log = logging.getLogger(__name__) RAW = Path("data/raw/bcra") # --------------------------------------------------------------- catálogo def catalogo(session, cfg) -> pd.DataFrame: url = cfg["bcra"]["base_url"] + cfg["bcra"]["catalogo"] js = get_json(session, url, timeout=cfg["general"]["request_timeout"]) df = pd.DataFrame(js.get("results", js.get("Results", []))) df.columns = [c[0].lower() + c[1:] for c in df.columns] # normaliza camelCase RAW.mkdir(parents=True, exist_ok=True) df.to_parquet(RAW / "catalogo_variables.parquet", index=False) log.info("Catálogo BCRA: %d variables", len(df)) return df def resolver_variables(cat: pd.DataFrame, patrones: dict[str, str]) -> pd.DataFrame: """Resuelve alias→idVariable por regex. Devuelve tabla de trazabilidad.""" filas = [] for alias, patron in patrones.items(): m = cat[cat["descripcion"].astype(str).str.contains(patron, regex=True, na=False)] if m.empty: log.warning("[%s] sin match para /%s/", alias, patron) filas.append({"alias": alias, "patron": patron, "idVariable": None, "descripcion": None, "estado": "NO_ENCONTRADA"}) continue # Si hay varios matches, tomamos el de descripción más corta (el más "general") best = m.assign(_len=m["descripcion"].str.len()).sort_values("_len").iloc[0] filas.append({ "alias": alias, "patron": patron, "idVariable": int(best["idVariable"]), "descripcion": best["descripcion"], "estado": "OK" if len(m) == 1 else f"AMBIGUA({len(m)})", }) res = pd.DataFrame(filas) res.to_parquet(RAW / "resolucion_variables.parquet", index=False) return res # --------------------------------------------------------------- series def serie(session, cfg, id_variable: int, desde: str, hasta: str) -> pd.DataFrame: """Descarga una serie completa paginando el endpoint v3.0.""" base = cfg["bcra"]["base_url"] + cfg["bcra"]["serie"].format(id=id_variable) limit = cfg["bcra"]["page_limit"] offset, frames = 0, [] while True: js = get_json(session, base, params={"desde": desde, "hasta": hasta, "limit": limit, "offset": offset}, timeout=cfg["general"]["request_timeout"]) rows = js.get("results", []) if not rows: break frames.append(pd.DataFrame(rows)) if len(rows) < limit: break offset += limit if offset > 500_000: # guardrail log.error("Paginación excesiva en idVariable=%s", id_variable) break if not frames: return pd.DataFrame(columns=["fecha", "valor", "idVariable"]) df = pd.concat(frames, ignore_index=True) df.columns = [c.lower() for c in df.columns] df["fecha"] = pd.to_datetime(df["fecha"], errors="coerce") df["valor"] = pd.to_numeric(df["valor"], errors="coerce") df["idvariable"] = id_variable return df.dropna(subset=["fecha"]).sort_values("fecha").reset_index(drop=True) def descargar_panel(session, cfg) -> pd.DataFrame: """Descarga todas las variables configuradas → panel largo tidy.""" cat = catalogo(session, cfg) res = resolver_variables(cat, cfg["bcra"]["variables"]) desde = cfg["general"]["fecha_inicio"] hasta = pd.Timestamp.today().strftime("%Y-%m-%d") out = [] for _, r in res[res["idVariable"].notna()].iterrows(): s = serie(session, cfg, int(r["idVariable"]), desde, hasta) if s.empty: log.warning("serie vacía: %s", r["alias"]); continue s["alias"] = r["alias"] s["descripcion"] = r["descripcion"] out.append(s) log.info("✓ %-26s n=%-6d [%s → %s]", r["alias"], len(s), s["fecha"].min().date(), s["fecha"].max().date()) panel = pd.concat(out, ignore_index=True) if out else pd.DataFrame() panel.to_parquet(RAW / "panel_monetario.parquet", index=False) return panel def cotizaciones(session, cfg, fecha: str | None = None) -> pd.DataFrame: url = cfg["bcra"]["base_url"] + cfg["bcra"]["cotizaciones"] js = get_json(session, url, params={"fecha": fecha} if fecha else None) res = js.get("results", {}) det = pd.DataFrame(res.get("detalle", [])) if not det.empty: det["fecha"] = pd.to_datetime(res.get("fecha")) return det ``` --- ### `src/extract/finanzas.py` ```python """Descubrimiento y descarga de planillas de deuda de la Secretaría de Finanzas.""" from __future__ import annotations import logging import re from pathlib import Path from urllib.parse import urljoin, urlparse import pandas as pd from bs4 import BeautifulSoup from ..common.http import download, register log = logging.getLogger(__name__) RAW = Path("data/raw/finanzas") def descubrir_archivos(session, cfg) -> pd.DataFrame: """Recorre las páginas de Finanzas y lista todos los datasets publicados.""" exts = tuple(cfg["finanzas"]["extensiones"]) clasif = cfg["finanzas"]["clasificador"] hits = [] for page in cfg["finanzas"]["paginas"]: try: r = session.get(page, timeout=cfg["general"]["request_timeout"]) r.raise_for_status() except Exception as e: # noqa: BLE001 log.error("No se pudo leer %s → %s", page, e) continue soup = BeautifulSoup(r.text, "lxml") for a in soup.find_all("a", href=True): href = urljoin(page, a["href"]) if not href.lower().split("?")[0].endswith(exts): continue texto = " ".join(a.get_text(" ", strip=True).split()) categoria = next( (k for k, pat in clasif.items() if re.search(pat, texto + " " + href)), "otros") hits.append({"pagina": page, "url": href, "texto": texto, "categoria": categoria, "archivo": Path(urlparse(href).path).name}) df = (pd.DataFrame(hits) .drop_duplicates(subset="url") .sort_values(["categoria", "archivo"]) .reset_index(drop=True)) RAW.mkdir(parents=True, exist_ok=True) df.to_csv(RAW / "_catalogo_archivos.csv", index=False) log.info("Finanzas: %d archivos descubiertos en %d páginas", len(df), df["pagina"].nunique() if len(df) else 0) return df def descargar(session, cfg, catalogo: pd.DataFrame, categorias: list[str] | None = None) -> pd.DataFrame: """Descarga los archivos de las categorías pedidas.""" sel = catalogo if categorias is None else catalogo[catalogo["categoria"].isin(categorias)] rutas = [] for _, r in sel.iterrows(): dest = RAW / r["categoria"] / r["archivo"] try: p = download(session, r["url"], dest, ttl_hours=cfg["general"]["cache_ttl_hours"]) rutas.append({**r.to_dict(), "local": str(p), "ok": True}) except Exception as e: # noqa: BLE001 log.error("fallo descarga %s → %s", r["url"], e) rutas.append({**r.to_dict(), "local": None, "ok": False, "error": str(e)}) return pd.DataFrame(rutas) # ------------------------------------------------------------------ inspect def inspeccionar(path: Path, n: int = 8) -> dict: """Modo --inspect: imprime hojas y encabezados para construir config/mapeos.yml.""" xl = pd.ExcelFile(path) info = {"archivo": str(path), "hojas": {}} for sh in xl.sheet_names: head = xl.parse(sh, header=None, nrows=n) info["hojas"][sh] = { "shape_preview": list(head.shape), "primeras_filas": head.fillna("").astype(str).values.tolist(), } return info ``` --- ### `src/extract/datos_gob.py` ```python """Series de Tiempo AR (apis.datos.gob.ar) con verificación/descubrimiento de IDs.""" from __future__ import annotations import logging from pathlib import Path import pandas as pd from ..common.http import get_json log = logging.getLogger(__name__) RAW = Path("data/raw/datosgob") def buscar(session, cfg, q: str, limit: int = 20) -> pd.DataFrame: url = cfg["datos_gob"]["base_url"] + "/search/" js = get_json(session, url, params={"q": q, "limit": limit}) df = pd.DataFrame(js.get("data", [])) cols = [c for c in ["id", "title", "description", "dataset_title", "frequency", "units", "start_date", "end_date"] if c in df.columns] return df[cols] if cols else df def verificar(session, cfg, ids: list[str]) -> list[str]: """Devuelve el subconjunto de IDs que la API acepta.""" ok = [] for sid in ids: try: get_json(session, cfg["datos_gob"]["base_url"] + "/series/", params={"ids": sid, "limit": 1}) ok.append(sid) except Exception: # noqa: BLE001 log.warning("ID inválido en datos.gob.ar: %s", sid) return ok def resolver(session, cfg) -> dict[str, str]: """Para cada alias: verifica candidatos; si fallan, busca por texto.""" out: dict[str, str] = {} for alias, spec in cfg["datos_gob"]["series"].items(): ids = verificar(session, cfg, spec.get("candidatos") or []) if ids: out[alias] = ids[0] log.info("✓ %-20s → %s (candidato)", alias, ids[0]) continue cand = buscar(session, cfg, spec["busqueda"], limit=10) if cand.empty: log.error("✗ %-20s sin resultados para «%s»", alias, spec["busqueda"]) continue out[alias] = cand.iloc[0]["id"] log.info("✓ %-20s → %s (búsqueda: %s)", alias, out[alias], cand.iloc[0].get("title", "")) return out def serie(session, cfg, ids: dict[str, str], start: str | None = None, collapse: str = "month") -> pd.DataFrame: """Descarga series y devuelve panel largo [fecha, alias, valor].""" if not ids: return pd.DataFrame(columns=["fecha", "alias", "valor"]) inv = {v: k for k, v in ids.items()} js = get_json(session, cfg["datos_gob"]["base_url"] + "/series/", params={ "ids": ",".join(ids.values()), "start_date": start or cfg["general"]["fecha_inicio"], "collapse": collapse, "collapse_aggregation": "avg", "format": "json", "limit": 5000, }) cols = [m["id"] for m in js["meta"][1:]] df = pd.DataFrame(js["data"], columns=["fecha"] + cols) df["fecha"] = pd.to_datetime(df["fecha"]) long = df.melt(id_vars="fecha", var_name="serie_id", value_name="valor") long["alias"] = long["serie_id"].map(inv) RAW.mkdir(parents=True, exist_ok=True) long.to_parquet(RAW / "series_macro.parquet", index=False) return long ``` --- ### `src/transform/clean.py` ```python """Normalización de la planilla maestra de deuda (Finanzas) a formato tidy.""" from __future__ import annotations import logging import re import unicodedata from pathlib import Path import numpy as np import pandas as pd log = logging.getLogger(__name__) INTERIM = Path("data/interim") # Diccionario canónico de columnas. El matching es difuso para tolerar # cambios de redacción entre publicaciones trimestrales. CANON = { "instrumento": [r"instrumento", r"t[ií]tulo", r"denominaci[oó]n", r"concepto"], "moneda": [r"^moneda", r"moneda\s+de\s+(emisi[oó]n|pago|origen)"], "legislacion": [r"legislaci[oó]n", r"ley\s+aplicable", r"jurisdicci[oó]n"], "acreedor": [r"acreedor", r"tenedor", r"tipo\s+de\s+acreedor"], "tipo_tasa": [r"tipo\s+de\s+tasa", r"clase\s+de\s+tasa"], "tasa": [r"^tasa", r"cup[oó]n", r"inter[eé]s\s+anual"], "f_emision": [r"fecha\s+de\s+emisi[oó]n", r"emisi[oó]n"], "f_vencimiento": [r"vencimiento", r"fecha\s+de\s+vto"], "vr_original": [r"valor\s+(nominal|residual)\s+original", r"monto\s+emitido"], "vr_residual": [r"valor\s+residual", r"saldo\s+(adeudado|de\s+capital)", r"stock"], "monto_usd": [r"(en\s+)?millones\s+de\s+(d[oó]lares|usd)", r"usd", r"d[oó]lares"], "monto_ars": [r"millones\s+de\s+pesos", r"\bars\b", r"\bpesos\b"], "fecha_corte": [r"fecha\s+de\s+corte", r"al\s+\d{2}/\d{2}/\d{4}"], } def _slug(s: str) -> str: s = unicodedata.normalize("NFKD", str(s)).encode("ascii", "ignore").decode() return re.sub(r"\s+", " ", s).strip().lower() def _match_cols(cols) -> dict[str, str]: out = {} for canon, patrones in CANON.items(): for c in cols: if any(re.search(p, _slug(c)) for p in patrones): out.setdefault(canon, c) break return out def _find_header_row(raw: pd.DataFrame, max_scan: int = 25) -> int: """Detecta la fila de encabezado: la que maximiza matches canónicos.""" best, best_score = 0, -1 for i in range(min(max_scan, len(raw))): score = len(_match_cols(raw.iloc[i].tolist())) if score > best_score: best, best_score = i, score log.debug("header row=%d score=%d", best, best_score) return best def _to_num(s: pd.Series) -> pd.Series: """Parsea números con formato es-AR (1.234.567,89) y paréntesis negativos.""" t = (s.astype(str) .str.replace(r"[^\d,.\-()]", "", regex=True) .str.replace(r"^$(.*)$$", r"-\1", regex=True) .str.replace(".", "", regex=False) .str.replace(",", ".", regex=False) .replace({"": np.nan, "-": np.nan})) return pd.to_numeric(t, errors="coerce") def normalizar_instrumentos(path: Path, sheet: str | int = 0, fecha_corte: str | None = None) -> pd.DataFrame: """Planilla de deuda por instrumento → tabla tidy canónica.""" raw = pd.read_excel(path, sheet_name=sheet, header=None, dtype=object) hrow = _find_header_row(raw) df = pd.read_excel(path, sheet_name=sheet, header=hrow) df = df.dropna(how="all").dropna(axis=1, how="all") mapping = _match_cols(df.columns) if "instrumento" not in mapping: raise ValueError( f"No se pudo identificar la columna 'instrumento' en {path.name}:{sheet}. " f"Columnas: {list(df.columns)[:25]} → mapeá manualmente en config/mapeos.yml") out = df.rename(columns={v: k for k, v in mapping.items()}) out = out[[c for c in CANON if c in out.columns]].copy() for c in ("vr_original", "vr_residual", "monto_usd", "monto_ars", "tasa"): if c in out: out[c] = _to_num(out[c]) for c in ("f_emision", "f_vencimiento"): if c in out: out[c] = pd.to_datetime(out[c], errors="coerce", dayfirst=True) # Elimina filas de subtotal/total (ruido clásico de estas planillas) patron_total = r"(?i)^\s*(total|subtotal|suma|resumen)" out = out[~out["instrumento"].astype(str).str.match(patron_total, na=False)] out = out[out["instrumento"].notna()] out["fecha_corte"] = pd.to_datetime(fecha_corte or out.get("fecha_corte"), errors="coerce") out["fuente_archivo"] = path.name # ---------------- enriquecimiento ---------------- out["moneda_grupo"] = np.where( out.get("moneda", pd.Series(dtype=object)).astype(str) .str.upper().str.strip().isin(["ARS", "$", "PESOS", "PESO", "ARP"]), "Moneda local", "Moneda extranjera") def _clasificar(i: str) -> str: i = _slug(i) if re.search(r"fmi|fondo monetario|dsg|derechos especiales", i): return "FMI" if re.search(r"bid|birf|caf|fonplata|banco mundial|organismo", i): return "Org. internacionales" if re.search(r"adelanto.*transitorio|letra.*intransferible", i): return "BCRA" if re.search(r"^letra|lede|lecap|lecer|lelink|\bletes\b", i): return "Letras" if re.search(r"^bon|^bono|boncer|bonar|global|discount|par|dual", i): return "Bonos" if re.search(r"pr[eé]stamo|club de par[ií]s|bilateral", i): return "Préstamos" return "Otros" out["familia"] = out["instrumento"].map(_clasificar) out["anio_vto"] = out["f_vencimiento"].dt.year out["plazo_residual_anios"] = ( (out["f_vencimiento"] - out["fecha_corte"]).dt.days / 365.25) INTERIM.mkdir(parents=True, exist_ok=True) out.to_parquet(INTERIM / f"deuda_instrumentos_{path.stem}.parquet", index=False) log.info("Normalizado %s:%s → %d instrumentos", path.name, sheet, len(out)) return out.reset_index(drop=True) def panel_bcra_a_mensual(panel: pd.DataFrame) -> pd.DataFrame: """Panel diario BCRA → mensual (último dato del mes y promedio).""" p = panel.copy() p["mes"] = p["fecha"].dt.to_period("M").dt.to_timestamp() g = (p.groupby(["alias", "mes"], as_index=False) .agg(valor_fdm=("valor", "last"), valor_prom=("valor", "mean"), n_obs=("valor", "size"))) return g.rename(columns={"mes": "fecha"}) ``` --- ### `src/transform/metrics.py` ```python """Métricas analíticas: descomposición de la variación, ratios, perfil de vencimientos.""" from __future__ import annotations import numpy as np import pandas as pd # ============================================================================= # 1. DESCOMPOSICIÓN DE LA VARIACIÓN DE LA DEUDA # ============================================================================= # D_t = Σ_i N_{i,t} · P_{i,t} · e_{i,t} # N = valor nominal residual P = coeficiente de ajuste (CER/UVA) e = TC # # ΔD = EmisiónNeta + EfectoFX + EfectoIndexación + Residual(cruzados) # # EmisiónNeta_t = Σ (N_t − N_{t−1}) · P_{t−1} · e_{t−1} # EfectoFX_t = Σ N_{t−1} · P_{t−1} · (e_t − e_{t−1}) # EfectoIndex_t = Σ N_{t−1} · (P_t − P_{t−1}) · e_{t−1} # Residual = ΔD − (los tres anteriores) ← términos de 2°/3° orden # ============================================================================= COMPONENTES = ["emision_neta", "efecto_fx", "efecto_indexacion", "residual"] def descomponer_variacion(df: pd.DataFrame, key: list[str] | None = None, col_fecha: str = "fecha_corte", col_n: str = "vr_residual", col_p: str = "coef_ajuste", col_e: str = "tc") -> pd.DataFrame: """ Descompone ΔDeuda entre cortes consecutivos. df debe tener, por instrumento y fecha de corte: - col_n : valor nominal residual (en moneda de emisión) - col_p : coeficiente de ajuste (1.0 si no ajusta) - col_e : tipo de cambio a USD (1.0 si ya está en USD) """ key = key or ["instrumento"] d = df.sort_values(key + [col_fecha]).copy() for c in (col_p, col_e): if c not in d: d[c] = 1.0 d[[col_p, col_e]] = d[[col_p, col_e]].fillna(1.0) g = d.groupby(key, sort=False) d["N0"], d["P0"], d["E0"] = g[col_n].shift(), g[col_p].shift(), g[col_e].shift() d["stock"] = d[col_n] * d[col_p] * d[col_e] d["stock0"] = d["N0"] * d["P0"] * d["E0"] d["emision_neta"] = (d[col_n] - d["N0"]) * d["P0"] * d["E0"] d["efecto_fx"] = d["N0"] * d["P0"] * (d[col_e] - d["E0"]) d["efecto_indexacion"] = d["N0"] * (d[col_p] - d["P0"]) * d["E0"] d["delta_total"] = d["stock"] - d["stock0"] d["residual"] = d["delta_total"] - d[COMPONENTES[:3]].sum(axis=1) agg = (d.groupby(col_fecha, as_index=False) .agg(stock=("stock", "sum"), delta_total=("delta_total", "sum"), **{c: (c, "sum") for c in COMPONENTES})) # Chequeo de consistencia contable (debe ser ~0) agg["check"] = agg["delta_total"] - agg[COMPONENTES].sum(axis=1) return agg # ============================================================================= # 2. RATIOS DE SOSTENIBILIDAD # ============================================================================= def ratios(deuda_usd: pd.Series, pib_usd: pd.Series, reservas_usd: pd.Series | None = None, intereses_usd: pd.Series | None = None, recaudacion_usd: pd.Series | None = None, exportaciones_usd: pd.Series | None = None) -> pd.DataFrame: out = pd.DataFrame({"deuda_usd": deuda_usd, "pib_usd": pib_usd}) out["deuda_pib_pct"] = 100 * out["deuda_usd"] / out["pib_usd"] if reservas_usd is not None: out["reservas_usd"] = reservas_usd out["deuda_reservas_x"] = out["deuda_usd"] / out["reservas_usd"] if intereses_usd is not None and recaudacion_usd is not None: out["intereses_recaudacion_pct"] = 100 * intereses_usd / recaudacion_usd if exportaciones_usd is not None: out["deuda_exportaciones_x"] = out["deuda_usd"] / exportaciones_usd return out def dinamica_deuda(d_pib_0: float, r: float, g: float, saldo_primario_pib: float, anios: int = 10, ajuste_fx_pib: float = 0.0) -> pd.DataFrame: """ Ecuación estándar de dinámica de deuda: d_t = d_{t-1} · (1+r)/(1+g) − sp_t + ajuste r = tasa real implícita, g = crecimiento real del PIB, sp = superávit primario/PIB. """ filas, d = [], d_pib_0 for t in range(1, anios + 1): snowball = d * (r - g) / (1 + g) d_new = d + snowball - saldo_primario_pib + ajuste_fx_pib filas.append({"t": t, "deuda_pib": d_new, "efecto_bola_nieve": snowball, "saldo_primario": -saldo_primario_pib, "variacion": d_new - d}) d = d_new return pd.DataFrame(filas) # ============================================================================= # 3. PERFIL DE VENCIMIENTOS Y VIDA PROMEDIO # ============================================================================= def perfil_vencimientos(df: pd.DataFrame, col_monto: str = "monto_usd", col_anio: str = "anio_vto", por: str | None = "moneda_grupo") -> pd.DataFrame: keys = [col_anio] + ([por] if por else []) p = (df.dropna(subset=[col_anio]) .groupby(keys, as_index=False)[col_monto].sum() .rename(columns={col_monto: "monto"})) p[col_anio] = p[col_anio].astype(int) return p.sort_values(keys) def vida_promedio(df: pd.DataFrame, col_monto: str = "monto_usd", col_plazo: str = "plazo_residual_anios") -> float: d = df.dropna(subset=[col_monto, col_plazo]) d = d[d[col_plazo] > 0] return float(np.average(d[col_plazo], weights=d[col_monto])) if len(d) else np.nan def concentracion_hhi(df: pd.DataFrame, por: str, col_monto: str = "monto_usd") -> float: """Herfindahl-Hirschman sobre la composición (0–10.000). Alto = concentrado.""" s = df.groupby(por)[col_monto].sum() share = (s / s.sum() * 100) ** 2 return float(share.sum()) # ============================================================================= # 4. DEUDA CONSOLIDADA (Tesoro + pasivos remunerados del BCRA) # ============================================================================= def deuda_consolidada(tesoro_usd: pd.Series, pasivos_bcra_ars: pd.Series, tc: pd.Series, intra_spn_usd: pd.Series | None = None ) -> pd.DataFrame: """ Consolidado ≈ Deuda bruta AC − deuda intra-SPN + pasivos remunerados BCRA (USD). Ver docs/metodologia.md §4 para los caveats (LEFI, doble contabilización). """ out = pd.DataFrame({ "deuda_tesoro_bruta_usd": tesoro_usd, "pasivos_remunerados_bcra_usd": pasivos_bcra_ars / tc, }) out["deuda_intra_spn_usd"] = (intra_spn_usd if intra_spn_usd is not None else 0.0) out["deuda_tesoro_mercado_usd"] = out["deuda_tesoro_bruta_usd"] - out["deuda_intra_spn_usd"] out["deuda_consolidada_usd"] = (out["deuda_tesoro_mercado_usd"] + out["pasivos_remunerados_bcra_usd"]) return out # ============================================================================= # 5. UTILIDADES DE REPORTE # ============================================================================= def variacion(s: pd.Series, periodos: int = 12) -> pd.Series: return s.pct_change(periodos) * 100 def cagr(s: pd.Series, anios: float) -> float: s = s.dropna() if len(s) < 2 or s.iloc[0] <= 0: return np.nan return ((s.iloc[-1] / s.iloc[0]) ** (1 / anios) - 1) * 100 ``` --- ### `src/load/warehouse.py` + `sql/` ```python """Carga a DuckDB y ejecución de los modelos SQL.""" from __future__ import annotations import logging from pathlib import Path import duckdb import pandas as pd log = logging.getLogger(__name__) DB = Path("data/processed/deuda.duckdb") SQLDIR = Path("sql") def build(tablas: dict[str, pd.DataFrame]) -> duckdb.DuckDBPyConnection: DB.parent.mkdir(parents=True, exist_ok=True) con = duckdb.connect(str(DB)) con.execute("CREATE SCHEMA IF NOT EXISTS raw;") con.execute("CREATE SCHEMA IF NOG EXISTS mart;".replace("NOG", "NOT")) for name, df in tablas.items(): con.register(f"_tmp_{name}", df) con.execute(f"CREATE OR REPLACE TABLE raw.{name} AS SELECT * FROM _tmp_{name};") log.info("raw.%-28s %7d filas × %d cols", name, *df.shape) for f in sorted(SQLDIR.glob("*.sql")): log.info("→ ejecutando %s", f.name) con.execute(f.read_text(encoding="utf-8")) return con def q(con, sql: str) -> pd.DataFrame: return con.execute(sql).fetchdf() ``` ```sql -- sql/03_marts.sql (extracto representativo) CREATE OR REPLACE VIEW mart.composicion_moneda AS SELECT fecha_corte, moneda_grupo, SUM(monto_usd) AS monto_usd, 100.0 * SUM(monto_usd) / SUM(SUM(monto_usd)) OVER (PARTITION BY fecha_corte) AS share_pct FROM raw.deuda_instrumentos GROUP BY 1, 2; CREATE OR REPLACE VIEW mart.perfil_vencimientos AS SELECT fecha_corte, anio_vto, moneda_grupo, familia, SUM(monto_usd) AS capital_usd FROM raw.deuda_instrumentos WHERE anio_vto IS NOT NULL GROUP BY 1, 2, 3, 4; CREATE OR REPLACE VIEW mart.kpis AS WITH ult AS (SELECT MAX(fecha_corte) AS f FROM raw.deuda_instrumentos) SELECT (SELECT f FROM ult) AS fecha_corte, SUM(monto_usd) AS deuda_bruta_usd, SUM(monto_usd) FILTER (WHERE moneda_grupo='Moneda extranjera') / NULLIF(SUM(monto_usd),0) * 100 AS pct_moneda_extranjera, SUM(monto_usd) FILTER (WHERE familia='FMI') AS deuda_fmi_usd, SUM(monto_usd) FILTER (WHERE anio_vto <= year((SELECT f FROM ult))+1) AS vto_12m_usd, SUM(monto_usd * plazo_residual_anios) / NULLIF(SUM(monto_usd),0) AS vida_promedio_anios, COUNT(*) AS n_instrumentos FROM raw.deuda_instrumentos WHERE fecha_corte = (SELECT f FROM ult); ``` --- ### `src/quality/checks.py` ```python """Validaciones de calidad con pandera + reporte HTML.""" from __future__ import annotations from pathlib import Path import pandas as pd import pandera as pa from pandera import Check, Column, DataFrameSchema OUT = Path("outputs/quality") schema_instrumentos = DataFrameSchema({ "instrumento": Column(str, nullable=False), "monto_usd": Column(float, Check.ge(0), nullable=True), "f_vencimiento": Column("datetime64[ns]", nullable=True), "fecha_corte": Column("datetime64[ns]", nullable=False), "moneda_grupo": Column(str, Check.isin(["Moneda local", "Moneda extranjera"])), "familia": Column(str, Check.isin( ["FMI", "Org. internacionales", "BCRA", "Letras", "Bonos", "Préstamos", "Otros"])), }, strict=False, coerce=True) schema_panel_bcra = DataFrameSchema({ "fecha": Column("datetime64[ns]", nullable=False), "alias": Column(str, nullable=False), "valor": Column(float, nullable=True), }, strict=False, coerce=True) def reglas_negocio(dfi: pd.DataFrame) -> pd.DataFrame: """Chequeos semánticos que pandera no cubre.""" r = [] def add(nombre, ok, detalle=""): r.append({"check": nombre, "estado": "PASS" if ok else "FAIL", "detalle": detalle}) add("Sin duplicados instrumento×corte", not dfi.duplicated(["instrumento", "fecha_corte"]).any(), f"{int(dfi.duplicated(['instrumento','fecha_corte']).sum())} duplicados") neg = int((dfi["monto_usd"] < 0).sum()) add("Montos no negativos", neg == 0, f"{neg} negativos") vto_pasado = int((dfi["f_vencimiento"] < dfi["fecha_corte"]).sum()) add("Vencimientos ≥ fecha de corte", vto_pasado == 0, f"{vto_pasado} instrumentos vencidos aún en stock (revisar: puede ser deuda en default)") cob = 100 * dfi["monto_usd"].notna().mean() add("Cobertura de monto_usd ≥ 95%", cob >= 95, f"{cob:.1f}%") out_fam = 100 * (dfi["familia"] == "Otros").mean() add("Clasificados en familia ≥ 90%", out_fam <= 10, f"{out_fam:.1f}% sin clasificar") return pd.DataFrame(r) def ejecutar(dfi: pd.DataFrame, panel: pd.DataFrame) -> pd.DataFrame: resultados = [] for nombre, df, sch in [("deuda_instrumentos", dfi, schema_instrumentos), ("panel_bcra", panel, schema_panel_bcra)]: try: sch.validate(df, lazy=True) resultados.append({"check": f"schema::{nombre}", "estado": "PASS", "detalle": ""}) except pa.errors.SchemaErrors as e: resultados.append({"check": f"schema::{nombre}", "estado": "FAIL", "detalle": f"{len(e.failure_cases)} fallas"}) res = pd.concat([pd.DataFrame(resultados), reglas_negocio(dfi)], ignore_index=True) OUT.mkdir(parents=True, exist_ok=True) res.to_html(OUT / "report.html", index=False) res.to_csv(OUT / "report.csv", index=False) return res ``` --- ### `src/viz/charts.py` ```python """Gráficos Plotly con doble salida: HTML interactivo y SVG para PDF.""" from __future__ import annotations import base64 from pathlib import Path import pandas as pd import plotly.express as px import plotly.graph_objects as go import plotly.io as pio FIGS = Path("outputs/figs") PALETA = ["#1f4e79", "#2e86ab", "#f6a623", "#c0392b", "#27ae60", "#8e44ad", "#7f8c8d", "#16a085"] TEMPLATE = go.layout.Template(layout=dict( font=dict(family="Inter, Segoe UI, system-ui, sans-serif", size=13, color="#1a1a1a"), colorway=PALETA, plot_bgcolor="white", paper_bgcolor="white", xaxis=dict(showgrid=False, linecolor="#d0d0d0", ticks="outside"), yaxis=dict(gridcolor="#ececec", zerolinecolor="#b0b0b0"), legend=dict(orientation="h", yanchor="bottom", y=1.02, x=0), margin=dict(l=60, r=24, t=56, b=52), hovermode="x unified", )) pio.templates["deuda"] = TEMPLATE pio.templates.default = "deuda" class Figura: """Wrapper que expone .html (div Plotly) y .svg_b64 (para WeasyPrint).""" def __init__(self, fig: go.Figure, slug: str, titulo: str, fuente: str, nota: str = ""): self.fig, self.slug, self.titulo = fig, slug, titulo self.fuente, self.nota = fuente, nota def html(self) -> str: return pio.to_html(self.fig, full_html=False, include_plotlyjs=False, div_id=f"fig-{self.slug}", config={"displaylogo": False, "responsive": True, "toImageButtonOptions": {"filename": self.slug}}) def img_src(self, width=1000, height=520, scale=2) -> str: FIGS.mkdir(parents=True, exist_ok=True) png = self.fig.to_image(format="png", width=width, height=height, scale=scale) (FIGS / f"{self.slug}.png").write_bytes(png) self.fig.write_image(FIGS / f"{self.slug}.svg", format="svg", width=width, height=height) return "data:image/png;base64," + base64.b64encode(png).decode() # ---------------------------------------------------------------- gráficos def g_stock_deuda(df: pd.DataFrame) -> Figura: fig = go.Figure() fig.add_bar(x=df["fecha_corte"], y=df["deuda_bruta_usd"] / 1000, name="Deuda bruta", marker_color=PALETA[0]) fig.add_scatter(x=df["fecha_corte"], y=df["deuda_pib_pct"], name="% del PIB", yaxis="y2", mode="lines+markers", line=dict(color=PALETA[3], width=3)) fig.update_layout( title="Stock de deuda bruta de la Administración Central", yaxis_title="Miles de millones de USD", yaxis2=dict(title="% del PIB", overlaying="y", side="right", showgrid=False, ticksuffix="%")) return Figura(fig, "stock-deuda", "Stock de deuda bruta", "Secretaría de Finanzas · INDEC (PIB)") def g_waterfall(desc: pd.DataFrame, periodo: str) -> Figura: labels = ["Stock inicial", "Emisión neta", "Efecto tipo de cambio", "Indexación (CER/UVA)", "Residual", "Stock final"] vals = [desc["stock0"], desc["emision_neta"], desc["efecto_fx"], desc["efecto_indexacion"], desc["residual"], desc["stock"]] fig = go.Figure(go.Waterfall( orientation="v", measure=["absolute", "relative", "relative", "relative", "relative", "total"], x=labels, y=[float(v) / 1000 for v in vals], text=[f"{float(v)/1000:,.1f}" for v in vals], connector=dict(line=dict(color="#999")), increasing=dict(marker_color=PALETA[3]), decreasing=dict(marker_color=PALETA[4]), totals=dict(marker_color=PALETA[0]))) fig.update_layout(title=f"Descomposición de la variación de la deuda · {periodo}", yaxis_title="Miles de millones de USD", hovermode="closest") return Figura(fig, f"waterfall-{periodo}", "Descomposición de la variación", "Elaboración propia sobre Finanzas y BCRA", "Residual = términos cruzados de 2.º orden (ver metodología §1).") def g_composicion(df: pd.DataFrame, dim: str, titulo: str) -> Figura: fig = px.area(df, x="fecha_corte", y="share_pct", color=dim, groupnorm=None, title=titulo) fig.update_layout(yaxis_title="% del total", yaxis_ticksuffix="%") fig.update_traces(hovertemplate="%{y:.1f}%") return Figura(fig, f"comp-{dim}", titulo, "Secretaría de Finanzas") def g_perfil_vencimientos(p: pd.DataFrame) -> Figura: fig = px.bar(p, x="anio_vto", y="monto", color="moneda_grupo", barmode="stack", title="Perfil de vencimientos de capital") fig.update_layout(xaxis_title="Año de vencimiento", yaxis_title="Millones de USD", xaxis=dict(dtick=1)) return Figura(fig, "perfil-vto", "Perfil de vencimientos", "Secretaría de Finanzas", "Excluye intereses y deuda elegible no regularizada.") def g_consolidada(df: pd.DataFrame) -> Figura: fig = go.Figure() for col, nombre in [("deuda_tesoro_mercado_usd", "Tesoro (con mercado)"), ("pasivos_remunerados_bcra_usd", "Pasivos remunerados BCRA")]: fig.add_bar(x=df.index, y=df[col] / 1000, name=nombre) fig.update_layout(barmode="stack", title="Deuda consolidada: Tesoro + cuasifiscal BCRA", yaxis_title="Miles de millones de USD") return Figura(fig, "consolidada", "Deuda consolidada", "Finanzas · BCRA", "Metodología de consolidación simplificada (§4).") def g_serie_bcra(panel: pd.DataFrame, aliases: list[str], titulo: str, slug: str, normalizar: bool = False) -> Figura: d = panel[panel["alias"].isin(aliases)].copy() if normalizar: d["valor"] = d.groupby("alias")["valor"].transform(lambda s: 100 * s / s.iloc[0]) fig = px.line(d, x="fecha", y="valor", color="alias", title=titulo) fig.update_layout(yaxis_title="Índice base 100" if normalizar else "Valor") return Figura(fig, slug, titulo, "BCRA · API Estadísticas v3.0") def g_heatmap_vto_mensual(df: pd.DataFrame) -> Figura: d = df.dropna(subset=["f_vencimiento"]).copy() d["anio"], d["mes"] = d["f_vencimiento"].dt.year, d["f_vencimiento"].dt.month piv = (d.pivot_table(index="anio", columns="mes", values="monto_usd", aggfunc="sum") .reindex(columns=range(1, 13))) fig = px.imshow(piv, aspect="auto", color_continuous_scale="Blues", labels=dict(color="MM USD"), title="Mapa de calor de vencimientos (año × mes)") fig.update_xaxes(tickmode="array", tickvals=list(range(1, 13)), ticktext=["E","F","M","A","M","J","J","A","S","O","N","D"]) return Figura(fig, "heatmap-vto", "Mapa de calor de vencimientos", "Secretaría de Finanzas") ``` --- ### `src/report/templates/informe.html.j2` ```jinja {{ meta.titulo }} — {{ meta.fecha_corte }} {% if mode == 'html' %}{% endif %}

Informe de Data Analytics · Finanzas Públicas

Movimientos y evolución del endeudamiento del Sector Público Nacional
Datos al {{ meta.fecha_corte }}

Período analizado: {{ meta.periodo }}
Fuentes primarias: BCRA · Secretaría de Finanzas · INDEC · datos.gob.ar
Generado: {{ meta.generado }}
Versión del pipeline: {{ meta.version }} · commit {{ meta.commit }}
Registros procesados: {{ "{:,}".format(meta.n_registros) }}
Validaciones: {{ meta.qa_ok }}/{{ meta.qa_total }} PASS

Documento de carácter analítico elaborado con información pública. No constituye asesoramiento financiero ni representa la posición oficial de ningún organismo.

01 Resumen ejecutivo

{% for k in kpis %}

{% endfor %}

Hallazgos principales

{{ h }}

{% for sec in secciones %}

{% if sec.intro %}

{% endif %} {% for fig in sec.figuras %}

Gráfico {{ fig.numero }}. {{ fig.titulo }}

{% endfor %} {% for tabla in sec.tablas %}

Fuente: {{ tabla.fuente }}

{% endfor %} {% if sec.lectura %}

Lectura analítica

{% endif %}

{% endfor %}

{{ "%02d"|format(secciones|length + 2) }} Metodología y limitaciones

Descomposición contable de la variación

Δ Deuda = Emisión neta + Efecto tipo de cambio + Efecto indexación + Residual

  Emisión neta   = Σᵢ (Nᵢ,ₜ − Nᵢ,ₜ₋₁) · Pᵢ,ₜ₋₁ · eᵢ,ₜ₋₁
  Efecto FX      = Σᵢ  Nᵢ,ₜ₋₁ · Pᵢ,ₜ₋₁ · (eᵢ,ₜ − eᵢ,ₜ₋₁)
  Efecto índice  = Σᵢ  Nᵢ,ₜ₋₁ · (Pᵢ,ₜ − Pᵢ,ₜ₋₁) · eᵢ,ₜ₋₁

  N = valor nominal residual · P = coeficiente de ajuste (CER/UVA) · e = tipo de cambio

Trazabilidad de fuentes

Resolución de variables del BCRA

Resultados de validación de calidad

Limitaciones conocidas

{{ l }}

``` ### `src/report/templates/styles.css` ```css :root{ --azul:#1f4e79; --azul2:#2e86ab; --amb:#f6a623; --rojo:#c0392b; --verde:#27ae60; --tinta:#1a1a1a; --gris:#5a6672; --linea:#e3e7eb; --fondo:#f7f9fb; } *{box-sizing:border-box} body{margin:0;font-family:Inter,"Segoe UI",system-ui,-apple-system,sans-serif; color:var(--tinta);line-height:1.6;font-size:15px;background:#fff} body.html{max-width:1180px;margin:0 auto;padding:0 28px 80px} /* ---------- impresión ---------- */ @page{size:A4;margin:19mm 16mm 20mm; @bottom-center{content:counter(page) " / " counter(pages); font-family:Inter,sans-serif;font-size:8.5pt;color:#8a949e} @bottom-left{content:"Deuda Pública Argentina — Informe analítico"; font-size:8pt;color:#8a949e} } @page:first{ @bottom-center{content:""} @bottom-left{content:""} } .pdf .plot{display:none} .pdf img{max-width:100%;height:auto} .pagebreak{break-before:page} figure,.kpi-grid,.tabla-wrap,.callout{break-inside:avoid} h2,h3{break-after:avoid} /* ---------- portada ---------- */ .portada{padding:54px 0 40px;border-bottom:4px solid var(--azul)} .kicker{text-transform:uppercase;letter-spacing:.16em;font-size:11px; font-weight:700;color:var(--azul2);margin:0 0 12px} h1{font-size:40px;line-height:1.12;margin:0 0 14px;color:var(--azul);letter-spacing:-.02em} .subtitulo{font-size:17px;color:var(--gris);margin:0 0 30px} .ficha{display:grid;grid-template-columns:repeat(3,1fr);gap:14px 26px;margin:0 0 26px} .ficha div{border-left:3px solid var(--linea);padding-left:12px} .ficha dt{font-size:10px;text-transform:uppercase;letter-spacing:.09em; color:var(--gris);font-weight:700;margin-bottom:3px} .ficha dd{margin:0;font-size:13.5px;font-weight:600} .ficha dd.ok{color:var(--verde)} .ficha dd.warn{color:var(--rojo)} .disclaimer{font-size:11.5px;color:var(--gris);background:var(--fondo); padding:12px 15px;border-radius:7px;border-left:3px solid var(--amb)} /* ---------- secciones ---------- */ section{padding:36px 0} h2{font-size:25px;color:var(--azul);margin:0 0 20px;display:flex; align-items:baseline;gap:13px;letter-spacing:-.01em} h2 .num{font-size:13px;font-weight:800;color:#fff;background:var(--azul2); padding:4px 9px;border-radius:5px;letter-spacing:.05em} h3{font-size:17.5px;color:var(--azul);margin:28px 0 11px} h4{font-size:14px;margin:20px 0 8px;color:var(--gris); text-transform:uppercase;letter-spacing:.07em} .intro{font-size:15.5px;color:var(--gris);max-width:72ch;margin-bottom:22px} /* ---------- KPIs ---------- */ .kpi-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(185px,1fr)); gap:13px;margin:22px 0 30px} .kpi{background:var(--fondo);border:1px solid var(--linea);border-radius:10px; padding:16px 15px;border-top:4px solid var(--azul2);display:flex;flex-direction:column;gap:3px} .kpi.alerta{border-top-color:var(--rojo)} .kpi.positivo{border-top-color:var(--verde)} .kpi.neutro{border-top-color:var(--gris)} .kpi-label{font-size:10.5px;text-transform:uppercase;letter-spacing:.08em; color:var(--gris);font-weight:700} .kpi-value{font-size:27px;font-weight:800;line-height:1.1;letter-spacing:-.02em} .kpi-delta{font-size:12.5px;font-weight:700;color:var(--azul2)} .kpi-note{font-size:10.5px;color:var(--gris);margin-top:2px} /* ---------- figuras y tablas ---------- */ figure{margin:28px 0 34px} .fig-titulo{font-size:14.5px;font-weight:700;margin-bottom:7px;color:var(--tinta)} .fig-fuente{font-size:10.5px;color:var(--gris);margin-top:7px; border-top:1px solid var(--linea);padding-top:5px} .plot{border:1px solid var(--linea);border-radius:9px;padding:6px;overflow:hidden} table{width:100%;border-collapse:collapse;font-size:11.5px;margin:9px 0} thead th{background:var(--azul);color:#fff;text-align:right;padding:8px 9px; font-weight:600;font-size:10.5px;text-transform:uppercase;letter-spacing:.04em} thead th:first-child{text-align:left} tbody td{padding:6px 9px;border-bottom:1px solid var(--linea);text-align:right; font-variant-numeric:tabular-nums} tbody td:first-child{text-align:left;font-weight:600} tbody tr:nth-child(even){background:#fafcfd} tbody tr:hover{background:#eef4f9} /* ---------- bloques de texto ---------- */ .callout{background:linear-gradient(180deg,#eff5fa,#f8fbfd);border:1px solid #cfe0ee; border-left:5px solid var(--azul);border-radius:9px;padding:18px 22px;margin:26px 0} .callout ol{margin:8px 0 0;padding-left:20px} .callout li{margin-bottom:9px} .lectura{background:var(--fondo);border-radius:9px;padding:16px 20px; margin-top:22px;border-left:4px solid var(--amb);font-size:14px} .formula{background:#1f2733;color:#e8eef5;padding:16px 19px;border-radius:9px; font-family:"JetBrains Mono",Consolas,monospace;font-size:11.5px; line-height:1.75;overflow-x:auto;white-space:pre} .limitaciones li{margin-bottom:8px;font-size:13.5px;color:var(--gris)} footer{border-top:3px solid var(--azul);padding:22px 0 10px;font-size:11px;color:var(--gris)} code{background:var(--fondo);padding:1px 5px;border-radius:3px;font-size:.92em} ``` --- ### `src/report/build_html.py` y `build_pdf.py` ```python """Ensamblado del informe (HTML interactivo y HTML-para-PDF).""" from __future__ import annotations import subprocess from datetime import datetime from pathlib import Path import pandas as pd from jinja2 import Environment, FileSystemLoader, select_autoescape TPL_DIR = Path(__file__).parent / "templates" VERSION = "1.0.0" def _commit() -> str: try: return subprocess.check_output(["git", "rev-parse", "--short", "HEAD"], text=True).strip() except Exception: # noqa: BLE001 return "n/d" def _env() -> Environment: return Environment(loader=FileSystemLoader(TPL_DIR), autoescape=select_autoescape(["html"]), trim_blocks=True, lstrip_blocks=True) def tabla_html(df: pd.DataFrame, formatos: dict | None = None, max_rows=40) -> str: d = df.head(max_rows).copy() return d.to_html(index=False, border=0, classes="tabla", escape=False, float_format=lambda x: f"{x:,.1f}", na_rep="—", formatters=formatos or {}) def render(contexto: dict, mode: str, destino: Path) -> Path: """mode ∈ {'html','pdf'} — cambia Plotly interactivo por imágenes embebidas.""" env = _env() tpl = env.get_template("informe.html.j2") css = (TPL_DIR / "styles.css").read_text(encoding="utf-8") # Materializa figuras según el modo secciones = [] n = 0 for sec in contexto["secciones"]: figs = [] for f in sec["figuras"]: n += 1 figs.append({ "numero": n, "titulo": f.titulo, "fuente": f.fuente, "nota": f.nota, "html": f.html() if mode == "html" else "", "img": f.img_src() if mode == "pdf" else "", }) secciones.append({**sec, "figuras": figs}) html = tpl.render(**{**contexto, "secciones": secciones, "mode": mode, "css": css}) destino.parent.mkdir(parents=True, exist_ok=True) destino.write_text(html, encoding="utf-8") return destino def meta_base(fecha_corte, periodo, n_registros, qa: pd.DataFrame) -> dict: ok = int((qa["estado"] == "PASS").sum()) return { "titulo": "Deuda Pública Argentina: movimientos y dinámica del endeudamiento", "fecha_corte": pd.Timestamp(fecha_corte).strftime("%d/%m/%Y"), "periodo": periodo, "generado": datetime.now().strftime("%d/%m/%Y %H:%M"), "version": VERSION, "commit": _commit(), "repo": "github.com//deuda-ar-analytics", "n_registros": int(n_registros), "qa_ok": ok, "qa_total": len(qa), "qa_pass": ok == len(qa), } ``` ```python # src/report/build_pdf.py """HTML → PDF. Motor principal: WeasyPrint. Fallback: Chromium headless.""" from __future__ import annotations import logging import shutil import subprocess from pathlib import Path log = logging.getLogger(__name__) def a_pdf(html_path: Path, pdf_path: Path) -> Path: pdf_path.parent.mkdir(parents=True, exist_ok=True) try: from weasyprint import HTML HTML(filename=str(html_path), base_url=str(html_path.parent)).write_pdf( str(pdf_path), presentational_hints=True) log.info("PDF generado con WeasyPrint → %s", pdf_path) return pdf_path except Exception as e: # noqa: BLE001 log.warning("WeasyPrint falló (%s). Probando Chromium headless…", e) for exe in ("chromium", "chromium-browser", "google-chrome", "msedge"): if shutil.which(exe): subprocess.run([exe, "--headless=new", "--disable-gpu", "--no-sandbox", f"--print-to-pdf={pdf_path}", "--print-to-pdf-no-header", "--no-pdf-header-footer", html_path.resolve().as_uri()], check=True) log.info("PDF generado con %s → %s", exe, pdf_path) return pdf_path raise RuntimeError( "No hay motor de PDF disponible. Instalá WeasyPrint " "(`pip install weasyprint` + libpango) o Chromium.") ``` --- ### `src/cli.py` — orquestador ```python """CLI del pipeline: python -m src.cli """ from __future__ import annotations import logging from datetime import datetime from pathlib import Path import pandas as pd import typer import yaml from rich.logging import RichHandler from .common.http import build_session, manifest_df from .extract import bcra, datos_gob, finanzas from .load import warehouse from .quality import checks from .report import build_html, build_pdf from .transform import clean, metrics from .viz import charts app = typer.Typer(add_completion=False, help="Pipeline de analítica de deuda pública AR") logging.basicConfig(level="INFO", format="%(message)s", handlers=[RichHandler(rich_tracebacks=True, show_path=False)]) log = logging.getLogger("deuda") CFG = yaml.safe_load(Path("config/fuentes.yml").read_text(encoding="utf-8")) SESSION = build_session(CFG["general"]["user_agent"], CFG["general"]["verify_ssl_bcra"]) @app.command() def discover(): """Descubre endpoints vivos e IDs de series; escribe config/resolved.yml.""" cat = bcra.catalogo(SESSION, CFG) res_bcra = bcra.resolver_variables(cat, CFG["bcra"]["variables"]) archivos = finanzas.descubrir_archivos(SESSION, CFG) ids = datos_gob.resolver(SESSION, CFG) Path("config/resolved.yml").write_text(yaml.safe_dump({ "generado": datetime.now().isoformat(timespec="seconds"), "bcra_variables": res_bcra.set_index("alias")["idVariable"].dropna().astype(int).to_dict(), "datos_gob_series": ids, "finanzas_archivos": archivos.to_dict("records"), }, allow_unicode=True, sort_keys=False), encoding="utf-8") print(res_bcra.to_string(index=False)) print(archivos[["categoria", "archivo"]].to_string(index=False)) log.info("→ config/resolved.yml actualizado. REVISALO antes de seguir.") @app.command() def extract(): """Descarga todos los datos crudos.""" bcra.descargar_panel(SESSION, CFG) cat = finanzas.descubrir_archivos(SESSION, CFG) finanzas.descargar(SESSION, CFG, cat, categorias=["deuda_instrumentos", "presentacion_grafica", "vencimientos"]) ids = yaml.safe_load(Path("config/resolved.yml").read_text())["datos_gob_series"] datos_gob.serie(SESSION, CFG, ids) log.info("EXTRACT OK — ver data/raw/_manifest.json") @app.command() def inspect(archivo: str): """Imprime hojas/encabezados de una planilla para mapearla.""" import json print(json.dumps(finanzas.inspeccionar(Path(archivo)), indent=2, ensure_ascii=False)) @app.command() def transform(): """Normaliza y calcula métricas.""" panel = pd.read_parquet("data/raw/bcra/panel_monetario.parquet") clean.panel_bcra_a_mensual(panel).to_parquet("data/interim/panel_bcra_mensual.parquet") archivos = sorted(Path("data/raw/finanzas/deuda_instrumentos").glob("*.xls*")) frames = [] for p in archivos: for sh in pd.ExcelFile(p).sheet_names: try: frames.append(clean.normalizar_instrumentos(p, sh)) except Exception as e: # noqa: BLE001 log.debug("skip %s:%s → %s", p.name, sh, e) if not frames: raise typer.Exit("No se pudo normalizar ninguna planilla. Usá `inspect` y mapeos.yml") dfi = pd.concat(frames, ignore_index=True) dfi.to_parquet("data/processed/deuda_instrumentos.parquet", index=False) log.info("TRANSFORM OK — %d filas", len(dfi)) @app.command() def report(pdf: bool = True): """Construye el informe HTML (y PDF).""" dfi = pd.read_parquet("data/processed/deuda_instrumentos.parquet") panel = pd.read_parquet("data/raw/bcra/panel_monetario.parquet") qa = checks.ejecutar(dfi, panel) con = warehouse.build({"deuda_instrumentos": dfi, "panel_bcra": panel}) kpis_sql = warehouse.q(con, "SELECT * FROM mart.kpis") comp_mon = warehouse.q(con, "SELECT * FROM mart.composicion_moneda ORDER BY 1") # --- figuras --- figs_comp = charts.g_composicion(comp_mon, "moneda_grupo", "Composición de la deuda por moneda") figs_perfil = charts.g_perfil_vencimientos( metrics.perfil_vencimientos(dfi[dfi.fecha_corte == dfi.fecha_corte.max()])) figs_heat = charts.g_heatmap_vto_mensual(dfi[dfi.fecha_corte == dfi.fecha_corte.max()]) figs_res = charts.g_serie_bcra(panel, ["reservas_internacionales"], "Reservas internacionales del BCRA", "reservas") k = kpis_sql.iloc[0] contexto = { "meta": build_html.meta_base(k["fecha_corte"], CFG["general"]["fecha_inicio"] + " → hoy", len(dfi), qa), "kpis": [ {"label": "Deuda bruta", "value": f"US$ {k.deuda_bruta_usd/1000:,.1f} MM", "delta": "", "note": "Administración Central", "tono": "neutro"}, {"label": "En moneda extranjera", "value": f"{k.pct_moneda_extranjera:.1f}%", "delta": "", "note": "share del stock", "tono": "alerta" if k.pct_moneda_extranjera > CFG["analisis"]["umbrales"]["fx_share_alerta"] else "positivo"}, {"label": "Vida promedio", "value": f"{k.vida_promedio_anios:.1f} años", "delta": "", "note": "ponderada por monto", "tono": "alerta" if k.vida_promedio_anios < CFG["analisis"]["umbrales"]["vida_promedio_min"] else "positivo"}, {"label": "Vencimientos 12m", "value": f"US$ {k.vto_12m_usd/1000:,.1f} MM", "delta": "", "note": "capital", "tono": "neutro"}, {"label": "Instrumentos", "value": f"{int(k.n_instrumentos):,}", "delta": "", "note": "en el stock", "tono": "neutro"}, ], "hallazgos": [ "Completar automáticamente con `src/report/insights.py` " "(generador de hallazgos basado en reglas sobre las métricas calculadas).", ], "secciones": [ {"titulo": "Stock y composición de la deuda", "intro": "Evolución del stock bruto y su estructura por moneda, " "legislación, tipo de instrumento y acreedor.", "figuras": [figs_comp], "tablas": [ {"titulo": "Composición por moneda (último corte)", "html": build_html.tabla_html(comp_mon[comp_mon.fecha_corte == comp_mon.fecha_corte.max()]), "fuente": "Secretaría de Finanzas"}], "lectura": ""}, {"titulo": "Perfil de vencimientos y riesgo de refinanciación", "intro": "Distribución temporal de los pagos de capital y concentración de vencimientos.", "figuras": [figs_perfil, figs_heat], "tablas": [], "lectura": ""}, {"titulo": "Contexto monetario y de reservas", "intro": "Variables del BCRA relevantes para la capacidad de pago.", "figuras": [figs_res], "tablas": [], "lectura": ""}, ], "metodologia": Path("docs/metodologia.md").read_text(encoding="utf-8") if Path("docs/metodologia.md").exists() else "", "tabla_manifest": build_html.tabla_html( manifest_df()[["url", "descargado_utc", "bytes", "sha256"]]), "tabla_resolucion": build_html.tabla_html( pd.read_parquet("data/raw/bcra/resolucion_variables.parquet")), "tabla_qa": build_html.tabla_html(qa), "limitaciones": [ "Las planillas de Finanzas cambian de estructura entre publicaciones; " "el normalizador es difuso y puede requerir mapeo manual (config/mapeos.yml).", "La deuda en default / no regularizada se informa según criterio oficial y " "puede no ser comparable entre cortes.", "La conversión a USD usa el TC de referencia A3500 de cierre de cada corte; " "otros tipos de cambio producen niveles distintos.", "El PIB en USD depende del TC elegido y de revisiones de INDEC; " "el ratio deuda/PIB es sensible a ambos.", "La consolidación con pasivos del BCRA es una aproximación: no elimina " "la totalidad de las tenencias cruzadas entre organismos públicos.", ], } html = build_html.render(contexto, "html", Path("outputs/html/informe_deuda.html")) log.info("HTML → %s", html) if pdf: pdf_html = build_html.render(contexto, "pdf", Path("outputs/html/_print.html")) build_pdf.a_pdf(pdf_html, Path("outputs/pdf/informe_deuda.pdf")) @app.command() def all(): """Pipeline completo.""" discover(); extract(); transform(); report() if __name__ == "__main__": app() ``` --- ### `Makefile` ```makefile PY := python -m src.cli .PHONY: setup discover extract transform report pdf all clean setup: python -m venv .venv && . .venv/bin/activate && pip install -U pip -r requirements.txt mkdir -p data/{raw,interim,processed} outputs/{html,pdf,figs,quality} discover: ; $(PY) discover # 1. resuelve IDs/URLs vivos → config/resolved.yml extract: ; $(PY) extract # 2. descarga datos crudos (+ manifest SHA256) transform: ; $(PY) transform # 3. normaliza y calcula métricas report: ; $(PY) report --no-pdf # 4a. informe HTML pdf: ; $(PY) report # 4b. informe HTML + PDF all: ; $(PY) all inspect: ; $(PY) inspect $(F) # make inspect F=data/raw/.../archivo.xlsx clean: rm -rf data/interim/* data/processed/* outputs/figs/* outputs/html/_print.html ``` --- ### `.github/workflows/refresh.yml` ```yaml name: Actualización mensual del informe de deuda on: schedule: [{ cron: "0 11 5 * *" }] # día 5 de cada mes, 08:00 ART workflow_dispatch: jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: { python-version: "3.12", cache: pip } - name: Dependencias del sistema (WeasyPrint) run: sudo apt-get update && sudo apt-get install -y libpango-1.0-0 libpangoft2-1.0-0 libcairo2 - run: pip install -r requirements.txt - run: make discover extract transform pdf - uses: actions/upload-artifact@v4 with: name: informe-deuda-${{ github.run_number }} path: | outputs/html/informe_deuda.html outputs/pdf/informe_deuda.pdf outputs/quality/report.html data/raw/_manifest.json - name: Publicar en GitHub Pages uses: peaceiris/actions-gh-pages@v4 with: github_token: ${{ secrets.GITHUB_TOKEN }} publish_dir: ./outputs/html ``` --- ## 6. Modelo de datos (diccionario) **`raw.deuda_instrumentos`** — grano: instrumento × fecha de corte | Campo | Tipo | Descripción | |---|---|---| | `instrumento` | text | Denominación oficial del título/préstamo | | `familia` | text | FMI · Org. internacionales · BCRA · Letras · Bonos · Préstamos · Otros | | `moneda` / `moneda_grupo` | text | Moneda de emisión / agrupación local-extranjera | | `legislacion` | text | Local o extranjera | | `acreedor` | text | Sector privado, intra-sector público, organismos | | `tipo_tasa` / `tasa` | text / float | Fija, variable, CER, dollar-linked / cupón nominal anual | | `f_emision`, `f_vencimiento` | date | Fechas de emisión y vencimiento final | | `vr_residual` | float | Valor nominal residual en moneda de emisión | | `monto_usd`, `monto_ars` | float | Stock valuado, en millones | | `fecha_corte` | date | Fecha de referencia del stock | | `plazo_residual_anios` | float | `(f_vencimiento − fecha_corte)/365.25` | | `fuente_archivo` | text | Archivo origen (join con `_manifest.json`) | **`raw.panel_bcra`** — grano: variable × fecha · `fecha`, `alias`, `idvariable`, `descripcion`, `valor` **Marts:** `mart.kpis`, `mart.composicion_moneda`, `mart.composicion_legislacion`, `mart.composicion_familia`, `mart.perfil_vencimientos`, `mart.descomposicion_variacion`, `mart.ratios_sostenibilidad`, `mart.deuda_consolidada`. --- ## 7. Secciones del informe generado | # | Sección | Visualizaciones | |---|---|---| | 01 | **Resumen ejecutivo** | 6 KPI cards + hallazgos automáticos | | 02 | **Stock y evolución** | Barras stock + línea % PIB · CAGR · variación i.a. | | 03 | **Anatomía del aumento** | **Waterfall de descomposición** (emisión neta / FX / indexación / residual) por período | | 04 | **Composición** | Área 100 % por moneda · legislación · familia · acreedor · HHI de concentración | | 05 | **Perfil de vencimientos** | Barras apiladas por año + mapa de calor año×mes + vida promedio + cobertura con reservas | | 06 | **Deuda consolidada y cuasifiscal** | Barras apiladas Tesoro + pasivos remunerados BCRA · reservas netas | | 07 | **Sostenibilidad** | Dinámica `d_t = d_{t−1}(1+r)/(1+g) − sp` · escenarios de sensibilidad (r, g, TC) | | 08 | **Metodología y limitaciones** | Fórmulas · manifest con SHA-256 · tabla de resolución de variables · resultados QA | --- ## 8. Cómo ejecutarlo ```bash git clone deuda-ar-analytics && cd deuda-ar-analytics make setup # PASO 1 — descubrimiento (¡revisá la salida!) make discover # → imprime: qué idVariable del BCRA matcheó cada alias, # qué XLSX publicó Finanzas, qué IDs de datos.gob.ar se resolvieron. # → escribe config/resolved.yml # PASO 2 — si alguna planilla no se normaliza automáticamente: make inspect F=data/raw/finanzas/deuda_instrumentos/archivo.xlsx # → copiá los nombres de hoja/columna a config/mapeos.yml # PASO 3 — pipeline make extract transform pdf # Resultados open outputs/html/informe_deuda.html open outputs/pdf/informe_deuda.pdf open outputs/quality/report.html # Exploración ad-hoc duckdb data/processed/deuda.duckdb -c "SELECT * FROM mart.kpis" ``` --- ## 9. Advertencias importantes (leelas antes de publicar) 1. **No incluí cifras numéricas en este diseño a propósito.** El pipeline las trae de la fuente en tiempo de ejecución. Nunca publiques un número de deuda que no hayas visto salir del `make extract` y contrastado con el PDF oficial del *Informe Mensual de la Deuda*. 2. **Validación cruzada obligatoria:** compará siempre tu total de deuda bruta contra (a) la *Presentación Gráfica de la Deuda* de Finanzas y (b) los informes de la **OPC**. Si difieren más de ~0,5 %, hay un problema de perímetro (Administración Central vs. SPN) o de doble conteo. 3. **Perímetro:** "Administración Central" ≠ "Sector Público Nacional" ≠ "deuda consolidada con BCRA". Declaralo explícitamente en cada gráfico. 4. **Intra-sector público:** una parte relevante de la deuda es con ANSES/FGS, BCRA y bancos públicos. Mostrar bruto y "con el sector privado + organismos" por separado cambia radicalmente la lectura. 5. **Valuación:** el nivel en USD depende del tipo de cambio usado; en períodos de brecha cambiaria, aclarar siempre cuál (A3500, oficial, implícito). 6. **TLS del BCRA:** si la API falla por certificado, no dejes `verify_ssl_bcra: false` en producción sin dejar constancia en el informe. 7. **Rate limiting:** el código ya incluye pausas y reintentos exponenciales. No lo saques: son APIs públicas gratuitas. --- ## 10. Extensiones sugeridas - **`src/report/insights.py`**: generador de hallazgos por reglas (ej.: *"el X % del aumento del período se explica por efecto tipo de cambio, no por emisión neta"*). - **Dashboard**: `streamlit run app.py` leyendo el mismo DuckDB. - **Backtest de proyecciones**: comparar el perfil de vencimientos publicado hace 12 meses contra lo efectivamente pagado. - **dbt-duckdb** para versionar los modelos SQL con tests y documentación. - **Comparación internacional**: World Bank IDS / FMI WEO API para benchmarking de deuda/PIB regional. - **Alertas**: webhook a Slack/Telegram cuando un KPI cruza un umbral de `config/fuentes.yml`. ¿Querés que desarrolle alguna pieza en profundidad — por ejemplo el generador automático de hallazgos, el módulo de escenarios de sostenibilidad, o los mapeos concretos de las planillas de Finanzas una vez que corras `make inspect`?

{{ meta.titulo }}

01 Resumen ejecutivo

Hallazgos principales

{{ "%02d"|format(loop.index + 1) }} {{ sec.titulo }}

{{ tabla.titulo }}

Lectura analítica

{{ "%02d"|format(secciones|length + 2) }} Metodología y limitaciones

Descomposición contable de la variación

Trazabilidad de fuentes

Resolución de variables del BCRA

Resultados de validación de calidad

Limitaciones conocidas