diff --git a/src/dados/carga_macro.py b/src/dados/carga_macro.py index 4aaf7fc..f77f890 100644 --- a/src/dados/carga_macro.py +++ b/src/dados/carga_macro.py @@ -7,7 +7,9 @@ Por padrão lê estes arquivos de data/raw/ (use as opções --euribor-12m, --euribor-3m, --selic, --selic-meta e --bc-credito para outros caminhos): - ECB_Euribor_Serie-Mensal-12-Meses_csv.csv Euribor 12 meses (BCE, série ...EURIBOR1YD_...) + ECB_Euribor_Serie-Mensal-12-Meses_csv.csv Euribor 12 meses (BCE, série ...EURIBOR1YD_...; + também aceita o nome sem o "_csv": + ECB_Euribor_Serie-Mensal-12-Meses.csv) ECB_Euribor_Serie-Mensal-3-Meses.csv Euribor 3 meses (BCE, série ...EURIBOR3MD_...) BC-Selic_Acumulada_Mes_Anuzalizada_base252.csv Selic acumulada no mês anualizada base 252 (Banco Central, SGS 4189, % ao ano) @@ -45,7 +47,10 @@ import numpy as np import pandas as pd -from carga_bondora import _carregar_env, _conectar, _copy +try: # rodando como script: python src/dados/carga_macro.py + from carga_bondora import _carregar_env, _conectar, _copy +except ModuleNotFoundError: # importado como módulo (testes): from src.dados import carga_macro + from src.dados.carga_bondora import _carregar_env, _conectar, _copy JANELA_INI = "2009-01" JANELA_FIM = "2021-07" @@ -53,6 +58,13 @@ MESES_PT = ["Jan", "Fev", "Mar", "Abr", "Mai", "Jun", "Jul", "Ago", "Set", "Out", "Nov", "Dez"] PASTA_PADRAO = Path("data/raw") NOMES_BC_20742 = ("bcdata.sgs.20742.csv", "bcdata_sgs_20742.csv") +# O download do BCE às vezes vem com "_csv" antes da extensão: os dois nomes valem. +NOMES_EURIBOR_12M = ("ECB_Euribor_Serie-Mensal-12-Meses_csv.csv", "ECB_Euribor_Serie-Mensal-12-Meses.csv") + + +def primeiro_existente(nomes: tuple[str, ...], pasta: Path = PASTA_PADRAO) -> Path | None: + """Caminho do primeiro nome que existe em `pasta`, na ordem dada; None se nenhum existir.""" + return next((pasta / n for n in nomes if (pasta / n).is_file()), None) # -------------------------------------------------------------------------- @@ -241,8 +253,8 @@ def main() -> None: ap.add_argument("--dsn", default=os.environ.get("DATABASE_URL"), help="padrão: variável de ambiente DATABASE_URL") ap.add_argument("--ddl", type=Path, help="arquivo .sql aplicado antes da carga") - ap.add_argument("--euribor-12m", type=Path, - default=PASTA_PADRAO / "ECB_Euribor_Serie-Mensal-12-Meses.csv") + ap.add_argument("--euribor-12m", type=Path, default=None, + help="padrão: data/raw/" + " ou ".join(NOMES_EURIBOR_12M)) ap.add_argument("--euribor-3m", type=Path, default=PASTA_PADRAO / "ECB_Euribor_Serie-Mensal-3-Meses.csv") ap.add_argument("--selic", type=Path, @@ -255,11 +267,14 @@ def main() -> None: a = ap.parse_args() if not a.dsn: sys.exit("ERRO: informe --dsn ou defina DATABASE_URL.") - obrigatorios = {"euribor_12m": a.euribor_12m, "euribor_3m": a.euribor_3m, "selic": a.selic} + # Sem --euribor-12m: o primeiro dos dois nomes que existir (se nenhum existir, o erro + # abaixo cita o primeiro nome, que é o do download atual do BCE). + e12 = a.euribor_12m or primeiro_existente(NOMES_EURIBOR_12M) or PASTA_PADRAO / NOMES_EURIBOR_12M[0] + obrigatorios = {"euribor_12m": e12, "euribor_3m": a.euribor_3m, "selic": a.selic} for nome, caminho in obrigatorios.items(): if not caminho.is_file(): sys.exit(f"ERRO: arquivo não encontrado: {caminho}") - bc = a.bc_credito or next((PASTA_PADRAO / n for n in NOMES_BC_20742 if (PASTA_PADRAO / n).is_file()), None) + bc = a.bc_credito or primeiro_existente(NOMES_BC_20742) if a.bc_credito is not None and not a.bc_credito.is_file(): sys.exit(f"ERRO: arquivo não encontrado: {a.bc_credito}") arquivos = dict(obrigatorios, diff --git a/tests/dados/test_carga_macro_nomes.py b/tests/dados/test_carga_macro_nomes.py new file mode 100644 index 0000000..e9a115b --- /dev/null +++ b/tests/dados/test_carga_macro_nomes.py @@ -0,0 +1,70 @@ +"""A carga macro aceita os dois nomes do arquivo da Euribor 12 meses (com e sem "_csv"). + +Não precisa de banco nem dos CSVs reais: os arquivos são criados vazios numa pasta +temporária e a função `carregar` é trocada por uma que só guarda o que recebeu. +""" +import sys + +import pytest + +from src.dados import carga_macro as cm + +COM_CSV, SEM_CSV = cm.NOMES_EURIBOR_12M +OUTROS = ("ECB_Euribor_Serie-Mensal-3-Meses.csv", "BC-Selic_Acumulada_Mes_Anuzalizada_base252.csv") + + +def _criar(pasta, *nomes): + pasta.mkdir(parents=True, exist_ok=True) + for nome in nomes: + (pasta / nome).write_text("", encoding="utf-8") + + +@pytest.fixture +def rodar_main(tmp_path, monkeypatch): + """Roda o main() dentro de tmp_path e devolve os arquivos que iriam para a carga.""" + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("DATABASE_URL", "postgresql://teste") + recebido = {} + monkeypatch.setattr(cm, "carregar", lambda dsn, ddl, arquivos: recebido.update(arquivos)) + + def _rodar(*args): + monkeypatch.setattr(sys, "argv", ["carga_macro.py", *args]) + cm.main() + return recebido + + return _rodar + + +def test_primeiro_existente_respeita_a_ordem(tmp_path): + _criar(tmp_path, COM_CSV, SEM_CSV) + assert cm.primeiro_existente(cm.NOMES_EURIBOR_12M, tmp_path) == tmp_path / COM_CSV + + +def test_primeiro_existente_sem_nenhum_devolve_none(tmp_path): + assert cm.primeiro_existente(cm.NOMES_EURIBOR_12M, tmp_path) is None + + +@pytest.mark.parametrize("nome", [COM_CSV, SEM_CSV]) +def test_main_acha_a_euribor_12m_pelos_dois_nomes(rodar_main, tmp_path, nome): + _criar(tmp_path / "data" / "raw", nome, *OUTROS) + arquivos = rodar_main() + assert arquivos["euribor_12m"].name == nome + + +def test_main_respeita_o_caminho_explicito(rodar_main, tmp_path): + _criar(tmp_path / "data" / "raw", COM_CSV, *OUTROS) + _criar(tmp_path / "outra", "euribor12.csv") + arquivos = rodar_main("--euribor-12m", "outra/euribor12.csv") + assert str(arquivos["euribor_12m"]).replace("\\", "/") == "outra/euribor12.csv" + + +def test_main_sem_euribor_12m_para_com_erro_claro(rodar_main, tmp_path): + _criar(tmp_path / "data" / "raw", *OUTROS) + with pytest.raises(SystemExit, match=COM_CSV): + rodar_main() + + +def test_bc_20742_continua_aceitando_os_dois_nomes(rodar_main, tmp_path): + _criar(tmp_path / "data" / "raw", SEM_CSV, *OUTROS, "bcdata_sgs_20742.csv") + arquivos = rodar_main() + assert arquivos["bc_credito"].name == "bcdata_sgs_20742.csv"