From ee933c42cd4e0d882641dbaf1e68cc39818f0ef5 Mon Sep 17 00:00:00 2001 From: Roberto Date: Tue, 16 Jun 2026 20:39:57 -0300 Subject: [PATCH 1/3] =?UTF-8?q?Add=20ANBIMA=20T=C3=ADtulos=20P=C3=BAblicos?= =?UTF-8?q?=20(TPF)=20secondary-market=20source?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New get_tpf() fetches daily reference rates for outstanding federal government bonds (LTN, LFT, NTN-B, NTN-C, NTN-F) from ANBIMA's public file www.anbima.com.br/informacoes/merc-sec/arqs/ms{ymd}.txt — no key, no auth, same @-separated layout as the existing debêntures feed. Wired through the stack mirroring get_debentures: - TPFQuote model + get_tpf() in sources/anbima/indices.py - GET /anbima/tpf (filter by titulo, limit) - findata anbima tpf CLI command - parse + endpoint tests; CHANGELOG entry Closes the one clean open gap on data.anbima.com.br's datasets that has a key-free equivalent. The portal's fund datasets remain reCAPTCHA-gated and are intentionally not automated. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 8 +++ src/findata/api/routers/anbima.py | 16 +++++ src/findata/cli.py | 37 +++++++++++ src/findata/sources/anbima/__init__.py | 4 ++ src/findata/sources/anbima/indices.py | 86 ++++++++++++++++++++++++++ tests/test_anbima.py | 60 ++++++++++++++++++ 6 files changed, 211 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b79950a..7357a0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ 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). + ## [0.3.1] — 2026-04-29 Patch release fixing 5 bugs caught in adversarial review of v0.3.0 by 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..75e96c9 100644 --- a/src/findata/sources/anbima/indices.py +++ b/src/findata/sources/anbima/indices.py @@ -32,6 +32,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 +90,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 +146,17 @@ def _date_to_iso(s: str) -> str: return f"{y}-{int(m):02d}-{int(d):02d}" +_COMPACT_DATE_LEN = 8 # YYYYMMDD + + +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(): + return f"{s[0:4]}-{s[4:6]}-{s[6:8]}" + 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. @@ -527,3 +559,57 @@ async def get_debentures(data_referencia: date | None = None) -> list[DebentureQ ) ) 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") + raw = await get_bytes(TPF_URL.format(ymd=ymd), cache_ttl=3600) + text = raw.decode("latin1", errors="replace") + 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: + if cells[0].strip().lower() == "titulo": + 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 + ), + ) + ) + return rows diff --git a/tests/test_anbima.py b/tests/test_anbima.py index c2d9911..dfce308 100644 --- a/tests/test_anbima.py +++ b/tests/test_anbima.py @@ -15,11 +15,13 @@ 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 _date_to_iso, _f_br, _ima_cache, get_debentures, get_ettj, + get_tpf, ) @@ -111,6 +113,40 @@ 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 + + # ── API smoke ──────────────────────────────────────────────────── @@ -150,6 +186,30 @@ 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_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() From 0be08daca7935ea7e159008e137ab015350797f8 Mon Sep 17 00:00:00 2001 From: Roberto Date: Tue, 16 Jun 2026 21:03:44 -0300 Subject: [PATCH 2/3] =?UTF-8?q?harden(anbima):=20make=20TPF/deb=C3=AAnture?= =?UTF-8?q?s=20fetch=20resilient=20per=20adversarial=20review?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review (grok-composer-2.5 external + host cross-validation) flagged that get_tpf shipped without the resilience get_ima_history already has. Fixes, all additive: - Weekend/holiday HTTP 404 now returns [] instead of surfacing as a 500. Shared helper _fetch_anbima_txt() catches HTTPStatusError, maps 404 -> [] (debug log), logs+re-raises other HTTP errors. Applied to get_debentures too, since it shared the same wart. - Download size capped (max_bytes=8 MiB) to bound pathological upstream bodies. - Header detection tolerates an accented "Título" via _norm_header() (NFKD accent strip); previously the exact == "titulo" gate would silently return [] on any accent drift. - _compact_to_iso() validates the YYYYMMDD it builds with date.fromisoformat; a shape-valid-but-impossible date (e.g. 20261399) keeps the raw token rather than fabricating "2026-13-99". - Warn-log when a file yields no recognisable header row. Tests: +5 negative-path cases (404->[] at lib and API layer, accented header, compact-date validation). 224 -> 229 tests; ruff + mypy clean. Deferred (whole-codebase, not TPF-specific; tracked separately): server-local date.today() default, shared-LRU cache namespacing, float->Decimal for PU. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 6 ++- src/findata/api/app.py | 2 +- src/findata/sources/anbima/indices.py | 55 ++++++++++++++++++++++++--- tests/test_anbima.py | 38 ++++++++++++++++++ 4 files changed, 93 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7357a0f..4aceedd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,11 @@ adheres to [Semantic Versioning](https://semver.org/). `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). + — 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. ## [0.3.1] — 2026-04-29 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/sources/anbima/indices.py b/src/findata/sources/anbima/indices.py index 75e96c9..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 @@ -149,11 +150,22 @@ def _date_to_iso(s: str) -> str: _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(): - return f"{s[0:4]}-{s[4:6]}-{s[6:8]}" + 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) @@ -516,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) ─────────────────────────────────────── @@ -523,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] = [] @@ -558,6 +595,8 @@ 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 @@ -577,8 +616,9 @@ async def get_tpf(data_referencia: date | None = None) -> list[TPFQuote]: """ d = data_referencia or date.today() ymd = d.strftime("%y%m%d") - raw = await get_bytes(TPF_URL.format(ymd=ymd), cache_ttl=3600) - text = raw.decode("latin1", errors="replace") + 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 @@ -586,7 +626,8 @@ async def get_tpf(data_referencia: date | None = None) -> list[TPFQuote]: if not cells or len(cells) < _TPF_MIN_COLS: continue if not header_seen: - if cells[0].strip().lower() == "titulo": + # Tolerate an accented header ("Título") — the file is latin-1. + if _norm_header(cells[0]).startswith("titul"): header_seen = True continue rows.append( @@ -612,4 +653,6 @@ async def get_tpf(data_referencia: date | None = None) -> list[TPFQuote]: ), ) ) + 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 dfce308..68e088e 100644 --- a/tests/test_anbima.py +++ b/tests/test_anbima.py @@ -16,6 +16,7 @@ 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, @@ -147,6 +148,34 @@ async def test_tpf_parses_at_separated_txt() -> None: 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 ──────────────────────────────────────────────────── @@ -199,6 +228,15 @@ def test_anbima_tpf_endpoint() -> None: 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( From 3222ec5b19341fff0f9d6de4fefc917de2efb844 Mon Sep 17 00:00:00 2001 From: Roberto Freitas <54923863+robertoecf@users.noreply.github.com> Date: Tue, 16 Jun 2026 22:29:29 -0300 Subject: [PATCH 3/3] Update src/findata/sources/anbima/indices.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> --- src/findata/sources/anbima/indices.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/findata/sources/anbima/indices.py b/src/findata/sources/anbima/indices.py index 0682303..d452588 100644 --- a/src/findata/sources/anbima/indices.py +++ b/src/findata/sources/anbima/indices.py @@ -626,8 +626,8 @@ async def get_tpf(data_referencia: date | None = None) -> list[TPFQuote]: 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"): + if not header_seen: + if cells[0].strip().lower().startswith("tit"): header_seen = True continue rows.append(