Add ANBIMA Títulos Públicos (TPF) secondary-market source

CHANGELOG.md

@@ -6,6 +6,18 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- **ANBIMA Títulos Públicos (TPF) secondary market** — `get_tpf()`,
+  `GET /anbima/tpf`, and `findata anbima tpf`. Daily reference rates for
+  outstanding federal government bonds (LTN, LFT, NTN-B, NTN-C, NTN-F) from
+  the public file `www.anbima.com.br/informacoes/merc-sec/arqs/ms{ymd}.txt`
+  — no key, no auth. Filterable by `titulo` (bond type). Hardened after
+  adversarial review: weekend/holiday HTTP 404 returns `[]` (not a 500),
+  accented header tolerated, compact dates validated, download size
+  capped, and failure boundaries logged. Same 404/size hardening applied
+  to the sibling `get_debentures` path.
+
 ### Fixed
 
 - **`cvm holdings` ignored bare-digit CNPJs.** The CDA reader compared the

src/findata/api/app.py

@@ -67,7 +67,7 @@ def _resolve_version() -> str:
     "openfinance": "Open Finance Brasil (public Directory + indicator Portal)",
     "b3": "B3 (official COTAHIST, indices, and optional stock quotes)",
     "yahoo": "Yahoo Finance chart endpoint (experimental, unofficial)",
-    "anbima": "ANBIMA (IMA family, ETTJ, debêntures — public file downloads)",
+    "anbima": "ANBIMA (IMA family, ETTJ, debêntures, TPF — public file downloads)",
     "receita": "Receita Federal (federal tax collection)",
     "aneel": "ANEEL (generation and transmission auctions)",
     "susep": "SUSEP (supervised insurance entities)",

src/findata/api/routers/anbima.py

@@ -46,3 +46,19 @@ async def debentures(
         needle = emissor.upper()
         rows = [r for r in rows if needle in r.emissor.upper()]
     return rows[:limit]
+
+
+@router.get("/tpf")
+async def tpf(
+    data: date | None = Query(default=None),
+    titulo: str | None = Query(
+        default=None, description="Filter by bond type, e.g. LTN, NTN-B, LFT"
+    ),
+    limit: int = Query(default=1000, ge=1, le=5000),
+) -> list[anbima.TPFQuote]:
+    """Daily secondary-market reference rates for federal government bonds (TPF)."""
+    rows = await anbima.get_tpf(data)
+    if titulo:
+        needle = titulo.upper()
+        rows = [r for r in rows if needle in r.titulo.upper()]
+    return rows[:limit]

src/findata/cli.py

@@ -766,6 +766,43 @@ def anbima_debentures(
     rprint(table)
 
 
+@anbima_app.command("tpf")
+def anbima_tpf(
+    d: str | None = typer.Option(None, "--date", "-d"),
+    titulo: str | None = typer.Option(None, "--titulo", "-t"),
+    n: int = typer.Option(50, "--last", "-n"),
+) -> None:
+    """Daily secondary-market reference rates for federal government bonds (TPF)."""
+    from findata.sources.anbima import get_tpf
+
+    dt = date.fromisoformat(d) if d else None
+    rows = _run(get_tpf(dt))
+    if titulo:
+        needle = titulo.upper()
+        rows = [r for r in rows if needle in r.titulo.upper()]
+    rows = rows[:n]
+    if not rows:
+        rprint("[yellow]No TPF quotes matched.[/yellow]")
+        return
+    table = Table(title=f"ANBIMA: Títulos Públicos ({dt or 'today'})")
+    table.add_column("Title", style="cyan")
+    table.add_column("Maturity", style="blue")
+    table.add_column("Rate Indicative %", style="green", justify="right")
+    table.add_column("Buy %", justify="right")
+    table.add_column("Sell %", justify="right")
+    table.add_column("PU R$", justify="right")
+    for r in rows:
+        table.add_row(
+            r.titulo,
+            r.data_vencimento,
+            _fmt(r.taxa_indicativa_pct, ".4f"),
+            _fmt(r.taxa_compra_pct, ".4f"),
+            _fmt(r.taxa_venda_pct, ".4f"),
+            _fmt(r.pu, ".2f"),
+        )
+    rprint(table)
+
+
 # ── IPEA commands ──────────────────────────────────────────────────
 
 ipea_app = typer.Typer(help="IPEA Data — macro series catalog", no_args_is_help=True)

src/findata/sources/anbima/__init__.py

@@ -20,19 +20,23 @@
     ETTJDataPoint,
     IMADataPoint,
     IMAFamily,
+    TPFQuote,
     get_debentures,
     get_ettj,
     get_ima,
     get_ima_history,
+    get_tpf,
 )
 
 __all__ = [
     "DebentureQuote",
     "ETTJDataPoint",
     "IMADataPoint",
     "IMAFamily",
+    "TPFQuote",
     "get_debentures",
     "get_ettj",
     "get_ima",
     "get_ima_history",
+    "get_tpf",
 ]

src/findata/sources/anbima/indices.py

@@ -13,6 +13,7 @@
 import logging
 import ssl
 import time
+import unicodedata
 from datetime import date, timedelta
 from enum import StrEnum
 from typing import Any
@@ -32,6 +33,7 @@
 IMA_HISTORY_DOWNLOAD_URL = "https://www.anbima.com.br/informacoes/ima/ima-sh-down.asp"
 ETTJ_URL = "https://www.anbima.com.br/informacoes/est-termo/CZ-down.asp"
 DEBENTURES_URL = "https://www.anbima.com.br/informacoes/merc-sec-debentures/arqs/db{ymd}.txt"
+TPF_URL = "https://www.anbima.com.br/informacoes/merc-sec/arqs/ms{ymd}.txt"
 
 
 # ── Models ────────────────────────────────────────────────────────
@@ -89,6 +91,26 @@ class DebentureQuote(BaseModel):
     referencia_ntn_b: str | None = None
 
 
+class TPFQuote(BaseModel):
+    """Secondary-market reference rates for a federal government bond (TPF)."""
+
+    data_referencia: str  # YYYY-MM-DD
+    titulo: str  # LTN, LFT, NTN-B, NTN-C, NTN-F, ...
+    codigo_selic: str
+    data_base_emissao: str  # YYYY-MM-DD
+    data_vencimento: str  # YYYY-MM-DD
+    taxa_compra_pct: float | None = None
+    taxa_venda_pct: float | None = None
+    taxa_indicativa_pct: float | None = None
+    pu: float | None = None
+    desvio_padrao: float | None = None
+    intervalo_ind_inf_d0_pct: float | None = None
+    intervalo_ind_sup_d0_pct: float | None = None
+    intervalo_ind_inf_d1_pct: float | None = None
+    intervalo_ind_sup_d1_pct: float | None = None
+    criterio: str | None = None
+
+
 # ── Helpers ───────────────────────────────────────────────────────
 
 
@@ -125,6 +147,28 @@ def _date_to_iso(s: str) -> str:
     return f"{y}-{int(m):02d}-{int(d):02d}"
 
 
+_COMPACT_DATE_LEN = 8  # YYYYMMDD
+
+
+def _norm_header(s: str) -> str:
+    """Lower-case and strip accents so 'Título' and 'Titulo' compare equal."""
+    nfkd = unicodedata.normalize("NFKD", s.strip().lower())
+    return "".join(c for c in nfkd if not unicodedata.combining(c))
+
+
+def _compact_to_iso(s: str) -> str:
+    """The TPF file ships dates as compact YYYYMMDD. Normalise to YYYY-MM-DD."""
+    s = s.strip()
+    if len(s) == _COMPACT_DATE_LEN and s.isdigit():
+        iso = f"{s[0:4]}-{s[4:6]}-{s[6:8]}"
+        try:
+            date.fromisoformat(iso)
+        except ValueError:
+            return s  # shape matched but not a real date — keep raw, don't fabricate
+        return iso
+    return _date_to_iso(s)
+
+
 # ── IMA (snapshot of the latest trading day) ─────────────────────
 # The file ANBIMA publishes is named `ima_completo.xls` but it's actually
 # a one-day snapshot of every IMA family — not a multi-year history.
@@ -484,15 +528,40 @@ async def get_ettj(data_referencia: date | None = None) -> list[ETTJDataPoint]:
     return rows
 
 
+# ── Daily @-separated TXT files (debêntures + TPF) ───────────────
+
+_HTTP_NOT_FOUND = 404
+_ANBIMA_TXT_MAX_BYTES = 8 * 1024 * 1024  # daily TXT is tiny; cap pathological bodies
+
+
+async def _fetch_anbima_txt(url: str, label: str, d: date) -> str | None:
+    """Fetch a daily ANBIMA ``@``-separated TXT file for date ``d``.
+
+    Returns the decoded text, or ``None`` when upstream has no file for that
+    day (weekends/holidays → HTTP 404). Other HTTP errors are logged and
+    re-raised so callers fail loudly rather than returning a silent empty.
+    """
+    try:
+        raw = await get_bytes(url, cache_ttl=3600, max_bytes=_ANBIMA_TXT_MAX_BYTES)
+    except httpx.HTTPStatusError as exc:
+        if exc.response.status_code == _HTTP_NOT_FOUND:
+            logger.debug("No %s file published for %s", label, d)
+            return None
+        logger.warning("%s fetch failed for %s: HTTP %s", label, d, exc.response.status_code)
+        raise
+    return raw.decode("latin1", errors="replace")
+
+
 # ── Debêntures (daily TXT) ───────────────────────────────────────
 
 
 async def get_debentures(data_referencia: date | None = None) -> list[DebentureQuote]:
     """Daily secondary-market quotes for outstanding debentures."""
     d = data_referencia or date.today()
     ymd = d.strftime("%y%m%d")
-    raw = await get_bytes(DEBENTURES_URL.format(ymd=ymd), cache_ttl=3600)
-    text = raw.decode("latin1", errors="replace")
+    text = await _fetch_anbima_txt(DEBENTURES_URL.format(ymd=ymd), "debentures", d)
+    if text is None:
+        return []
     iso = d.strftime("%Y-%m-%d")
     reader = csv.reader(io.StringIO(text), delimiter="@")
     rows: list[DebentureQuote] = []
@@ -526,4 +595,64 @@ async def get_debentures(data_referencia: date | None = None) -> list[DebentureQ
                 ),
             )
         )
+    if not header_seen:
+        logger.warning("debentures file for %s had no recognisable header row", d)
+    return rows
+
+
+# ── Títulos Públicos Federais (daily TXT) ────────────────────────
+
+_TPF_MIN_COLS = 14
+_TPF_CRITERIO_IDX = 14
+
+
+async def get_tpf(data_referencia: date | None = None) -> list[TPFQuote]:
+    """Daily secondary-market reference rates for federal government bonds (TPF).
+
+    The same `@`-separated layout as the debêntures file: a title line, a
+    blank line, a header row starting with ``Titulo``, then one row per
+    outstanding bond (LTN, LFT, NTN-B, NTN-C, NTN-F). Dates arrive compact
+    (``YYYYMMDD``); rates are Brazilian-formatted decimals.
+    """
+    d = data_referencia or date.today()
+    ymd = d.strftime("%y%m%d")
+    text = await _fetch_anbima_txt(TPF_URL.format(ymd=ymd), "TPF", d)
+    if text is None:
+        return []
+    reader = csv.reader(io.StringIO(text), delimiter="@")
+    rows: list[TPFQuote] = []
+    header_seen = False
+    for cells in reader:
+        if not cells or len(cells) < _TPF_MIN_COLS:
+            continue
+        if not header_seen:
+            # Tolerate an accented header ("Título") — the file is latin-1.
+            if _norm_header(cells[0]).startswith("titul"):
+                header_seen = True
+            continue
+        rows.append(
+            TPFQuote(
+                titulo=cells[0].strip(),
+                data_referencia=_compact_to_iso(cells[1]),
+                codigo_selic=cells[2].strip(),
+                data_base_emissao=_compact_to_iso(cells[3]),
+                data_vencimento=_compact_to_iso(cells[4]),
+                taxa_compra_pct=_f_br(cells[5]),
+                taxa_venda_pct=_f_br(cells[6]),
+                taxa_indicativa_pct=_f_br(cells[7]),
+                pu=_f_br(cells[8]),
+                desvio_padrao=_f_br(cells[9]),
+                intervalo_ind_inf_d0_pct=_f_br(cells[10]),
+                intervalo_ind_sup_d0_pct=_f_br(cells[11]),
+                intervalo_ind_inf_d1_pct=_f_br(cells[12]),
+                intervalo_ind_sup_d1_pct=_f_br(cells[13]),
+                criterio=(
+                    (cells[_TPF_CRITERIO_IDX].strip() or None)
+                    if len(cells) > _TPF_CRITERIO_IDX
+                    else None
+                ),
+            )
+        )
+    if not header_seen:
+        logger.warning("TPF file for %s had no recognisable header row", d)
     return rows