diff --git a/CHANGELOG.md b/CHANGELOG.md index e263276..2af4a22 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/findata/api/app.py b/src/findata/api/app.py index 56f53fc..f14fbdf 100644 --- a/src/findata/api/app.py +++ b/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)", diff --git a/src/findata/api/routers/anbima.py b/src/findata/api/routers/anbima.py index efc7705..a92346c 100644 --- a/src/findata/api/routers/anbima.py +++ b/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] diff --git a/src/findata/cli.py b/src/findata/cli.py index f4b6871..c998e53 100644 --- a/src/findata/cli.py +++ b/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) diff --git a/src/findata/sources/anbima/__init__.py b/src/findata/sources/anbima/__init__.py index 5923572..c5463dc 100644 --- a/src/findata/sources/anbima/__init__.py +++ b/src/findata/sources/anbima/__init__.py @@ -20,10 +20,12 @@ ETTJDataPoint, IMADataPoint, IMAFamily, + TPFQuote, get_debentures, get_ettj, get_ima, get_ima_history, + get_tpf, ) __all__ = [ @@ -31,8 +33,10 @@ "ETTJDataPoint", "IMADataPoint", "IMAFamily", + "TPFQuote", "get_debentures", "get_ettj", "get_ima", "get_ima_history", + "get_tpf", ] diff --git a/src/findata/sources/anbima/indices.py b/src/findata/sources/anbima/indices.py index 8a3419c..0682303 100644 --- a/src/findata/sources/anbima/indices.py +++ b/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,6 +528,30 @@ 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) ─────────────────────────────────────── @@ -491,8 +559,9 @@ async def get_debentures(data_referencia: date | None = None) -> list[DebentureQ """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 diff --git a/tests/test_anbima.py b/tests/test_anbima.py index c2d9911..68e088e 100644 --- a/tests/test_anbima.py +++ b/tests/test_anbima.py @@ -15,11 +15,14 @@ from findata.sources.anbima.indices import ( DEBENTURES_URL, # noqa: F401 — exported constant, helps type-checking ETTJ_URL, + TPF_URL, # noqa: F401 — exported constant, helps type-checking + _compact_to_iso, _date_to_iso, _f_br, _ima_cache, get_debentures, get_ettj, + get_tpf, ) @@ -111,6 +114,68 @@ async def test_debentures_parses_at_separated_txt() -> None: assert out[1].taxa_compra_pct is None # "--" +# ── Títulos Públicos (TPF) ──────────────────────────────────────── + + +_TPF_TXT = ( + "ANBIMA - Associação ...\n" + "\n" + "Titulo@Data Referencia@Codigo SELIC@Data Base/Emissao@Data Vencimento@" + "Tx. Compra@Tx. Venda@Tx. Indicativas@PU@Desvio padrao@" + "Interv. Ind. Inf. (D0)@Interv. Ind. Sup. (D0)@" + "Interv. Ind. Inf. (D+1)@Interv. Ind. Sup. (D+1)@Criterio\n" + "LTN@20260612@100000@20230106@20260701@14,3812@14,3519@14,3671@993,098676@" + "0,00641938699644@14,2321@14,5625@14,222@14,5595@Calculado\n" + "LFT@20260612@210100@20000701@20260901@0,0016@-0,0034@0,0006@19201,839214@" + "0,00172517877914@-0,0453@0,032@-0,0468@0,0303@Calculado\n" +) + + +@respx.mock +async def test_tpf_parses_at_separated_txt() -> None: + respx.get(re.compile(r"https://.*ms\d{6}\.txt")).mock( + return_value=httpx.Response(200, text=_TPF_TXT, headers={"Content-Type": "text/plain"}) + ) + out = await get_tpf(date(2026, 6, 12)) + assert len(out) == 2 + assert out[0].titulo == "LTN" + assert out[0].data_referencia == "2026-06-12" + assert out[0].data_vencimento == "2026-07-01" # compact YYYYMMDD → ISO + assert out[0].taxa_indicativa_pct == pytest.approx(14.3671) + assert out[0].pu == pytest.approx(993.098676) + assert out[0].criterio == "Calculado" + assert out[1].titulo == "LFT" + assert out[1].taxa_venda_pct == pytest.approx(-0.0034) # negative rate kept + + +def test_compact_to_iso_validates_dates() -> None: + assert _compact_to_iso("20260701") == "2026-07-01" + # Shape matches but it isn't a real date — keep the raw token, don't fabricate. + assert _compact_to_iso("20261399") == "20261399" + # Non-compact inputs fall through to the DD/MM/YYYY normaliser. + assert _compact_to_iso("01/07/2026") == "2026-07-01" + + +@respx.mock +async def test_tpf_weekend_404_returns_empty() -> None: + """A missing daily file (weekend/holiday → HTTP 404) yields [], not a raise.""" + respx.get(re.compile(r"https://.*ms\d{6}\.txt")).mock(return_value=httpx.Response(404)) + out = await get_tpf(date(2026, 6, 13)) # a Saturday + assert out == [] + + +@respx.mock +async def test_tpf_tolerates_accented_header() -> None: + """An accented 'Título' header must still be recognised (no silent empty).""" + # The real file is latin-1; send raw latin-1 bytes so "í" survives decoding. + accented = _TPF_TXT.replace("Titulo@", "Título@", 1).encode("latin1") + respx.get(re.compile(r"https://.*ms\d{6}\.txt")).mock( + return_value=httpx.Response(200, content=accented, headers={"Content-Type": "text/plain"}) + ) + out = await get_tpf(date(2026, 6, 12)) + assert [r.titulo for r in out] == ["LTN", "LFT"] + + # ── API smoke ──────────────────────────────────────────────────── @@ -150,6 +215,39 @@ def test_anbima_debentures_filter_by_emissor() -> None: assert [d["codigo"] for d in r.json()] == ["VALE13"] +@respx.mock +def test_anbima_tpf_endpoint() -> None: + respx.get(re.compile(r"https://.*ms\d{6}\.txt")).mock( + return_value=httpx.Response(200, text=_TPF_TXT, headers={"Content-Type": "text/plain"}) + ) + client = TestClient(app) + r = client.get("/anbima/tpf?data=2026-06-12") + assert r.status_code == 200 + body = r.json() + assert len(body) == 2 + assert body[0]["titulo"] == "LTN" + + +@respx.mock +def test_anbima_tpf_endpoint_404_is_empty_not_500() -> None: + respx.get(re.compile(r"https://.*ms\d{6}\.txt")).mock(return_value=httpx.Response(404)) + client = TestClient(app) + r = client.get("/anbima/tpf?data=2026-06-13") + assert r.status_code == 200 + assert r.json() == [] + + +@respx.mock +def test_anbima_tpf_filter_by_titulo() -> None: + respx.get(re.compile(r"https://.*ms\d{6}\.txt")).mock( + return_value=httpx.Response(200, text=_TPF_TXT, headers={"Content-Type": "text/plain"}) + ) + client = TestClient(app) + r = client.get("/anbima/tpf?data=2026-06-12&titulo=LFT") + assert r.status_code == 200 + assert [d["titulo"] for d in r.json()] == ["LFT"] + + def test_root_endpoint_lists_anbima_in_main_sources() -> None: client = TestClient(app) body = client.get("/meta").json()