"""Etapa 3c: validação dos sinais em um período independente (2015–2019). Os sinais foram selecionados com os óbitos de 2020–2024. Aqui testamos os MESMOS municípios e grupos em 2015–2019, período que não foi usado na seleção. Nada é reselecionado: não há novos critérios geográficos nem novos limiares, só o teste dos pares município–grupo já definidos. CRITÉRIOS DEFINIDOS ANTES DE VER OS DADOS DE 2015–2019 (registrados no git antes da execução): Critério principal: um sinal "replica" se, em 2015–2019, a RMP sem suavização em relação ao próprio estado for maior que 1 e o teste de Poisson unilateral tiver p < 0,05. Resumo: proporção de sinais que replicam, comparada com os 5% esperados se não houvesse excesso real (teste binomial unilateral). Calculado para: (a) os 45 sinais após os critérios geográficos; (b) os 145 sinais; (c) os 16 sinais de câncer de esôfago. Secundários (descritivos, sem limiar): direção do efeito (RMP > 1, com qualquer p); RMP agrupada dos sinais de esôfago por região (Minas Gerais, Sul e Bahia), com IC 95% exato; razão entre a RMP de 2015–2019 e a de 2020–2024 (espera-se menor que 1 por regressão à média). População de 2015–2019: interpolação linear, por município, sexo e faixa etária, entre o Censo 2010 e o Censo 2022, no ano central (2017); pessoa-anos = 5 × população de 2017. Municípios criados depois de 2010 usam a população de 2022. Uso: python validar.py (depois de baixar.py validacao, preparar.py validacao e analisar.py) """ import numpy as np import pandas as pd from scipy import stats from analisar import carregar, comparar from config import ANOS_VALIDACAO, BRUTOS, GRUPOS, PROCESSADOS, RESULTADOS ALFA = 0.05 def populacao_2017(pop22): p10 = pd.read_csv(BRUTOS / "populacao_censo2010.csv").rename(columns={"pop": "pop10"}) p = pop22.merge(p10, on=["cod_ibge", "sexo", "faixa"], how="left") sem_2010 = p["pop10"].isna() p["pop"] = np.where(sem_2010, p["pop"], p["pop10"] + (7 / 12) * (p["pop"] - p["pop10"])).clip(min=0) p["pessoa_anos"] = p["pop"] * len(ANOS_VALIDACAO) n_sem = p.loc[sem_2010, "cod_ibge"].nunique() return p.drop(columns="pop10"), n_sem def ic_exato(o, e): inf = stats.chi2.ppf(0.025, 2 * o) / 2 / e if o > 0 else 0.0 sup = stats.chi2.ppf(0.975, 2 * (o + 1)) / 2 / e return inf, sup def binomial(k, n, p=ALFA): return stats.binomtest(k, n, p, alternative="greater").pvalue if n else np.nan def main(): pop22, _, _ = carregar() pop, n_sem = populacao_2017(pop22) ob = pd.read_csv(PROCESSADOS / f"obitos_agregados_{ANOS_VALIDACAO[0]}_{ANOS_VALIDACAO[-1]}.csv") ob["uf"] = ob["cod6"] // 10000 ob = ob[ob["cod6"].isin(set(pop["cod6"]))] base_idx = pd.Index(sorted(pop["cod6"].unique()), name="cod6") disc = pd.read_csv(RESULTADOS / "municipios_rmp.csv") sinais = disc[disc["sinal"]].copy() linhas = [] for grupo, info in GRUPOS.items(): ob_g = ob[ob["grupo"] == grupo] pop_g = pop if info["sexo"] is None else pop[pop["sexo"] == info["sexo"]] ob_g = ob_g if info["sexo"] is None else ob_g[ob_g["sexo"] == info["sexo"]] obs = ob_g.groupby("cod6")["obitos"].sum().reindex(base_idx).fillna(0) uf = comparar(obs, pop_g, ob_g, "uf") br = comparar(obs, pop_g, ob_g, None) d = pd.DataFrame({ "grupo": grupo, "obs_val": obs.astype(int), "esp_uf_val": uf["esperados"], "rmp_uf_val": uf["rmp"], "p_uf_val": uf["p_acima"], "rmp_suav_uf_val": uf["rmp_suav"], "esp_br_val": br["esperados"], "rmp_br_val": br["rmp"], }).reset_index() linhas.append(d) val = pd.concat(linhas, ignore_index=True) s = sinais.merge(val, on=["cod6", "grupo"], how="left") s["replica"] = (s["rmp_uf_val"] > 1) & (s["p_uf_val"] < ALFA) s["mesma_direcao"] = s["rmp_uf_val"] > 1 s["razao_val_disc"] = s["rmp_uf_val"] / s["rmp_uf"] ics = [ic_exato(o, e) for o, e in zip(s["obs_val"], s["esp_uf_val"])] s["ic95_inf_uf_val"], s["ic95_sup_uf_val"] = zip(*ics) s["nome_grupo"] = s["grupo"].map(lambda g: GRUPOS[g]["nome"]) cols = ["contexto", "nome_grupo", "grupo", "municipio", "uf_sigla", "cod_ibge", "observados", "esperados_uf", "rmp_uf", "rmp_suav_uf", "obs_val", "esp_uf_val", "rmp_uf_val", "ic95_inf_uf_val", "ic95_sup_uf_val", "p_uf_val", "rmp_suav_uf_val", "rmp_br_val", "replica", "mesma_direcao", "razao_val_disc"] s = s.sort_values(["contexto", "grupo", "rmp_uf_val"], ascending=[True, True, False]) s[cols].to_csv(RESULTADOS / "validacao_sinais.csv", index=False, float_format="%.4g") conjuntos = { "a_apos_criterios_geograficos": s[s["contexto"] == "prioritário"], "b_todos_os_sinais": s, "c_esofago_apos_criterios": s[(s["grupo"] == "esofago") & (s["contexto"] == "prioritário")], } resumo = [] for nome, d in conjuntos.items(): n, k = len(d), int(d["replica"].sum()) resumo.append({"conjunto": nome, "n": n, "replicam": k, "pct_replicam": 100 * k / n if n else np.nan, "esperado_sem_excesso": ALFA * n, "p_binomial": binomial(k, n), "mesma_direcao": int(d["mesma_direcao"].sum()), "mediana_razao_val_disc": d["razao_val_disc"].median()}) # RMP agrupada dos sinais de esôfago, por região esof = conjuntos["c_esofago_apos_criterios"] regioes = {"Minas Gerais": ["MG"], "Sul (PR, SC, RS)": ["PR", "SC", "RS"], "Bahia": ["BA"]} for reg, ufs in regioes.items(): d = esof[esof["uf_sigla"].isin(ufs)] o, e = d["obs_val"].sum(), d["esp_uf_val"].sum() inf, sup = ic_exato(o, e) resumo.append({"conjunto": f"esofago_agrupado_{reg}", "n": len(d), "obs_val": int(o), "esp_uf_val": e, "rmp_agrupada_val": o / e, "ic95_inf": inf, "ic95_sup": sup, "p_poisson": stats.poisson.sf(o - 1, e)}) resumo = pd.DataFrame(resumo) resumo.to_csv(RESULTADOS / "validacao_resumo.csv", index=False, float_format="%.4g") pd.set_option("display.width", 250, "display.max_columns", 30) print(f"Municípios sem Censo 2010 (usam 2022): {n_sem}") print(f"Óbitos por câncer 2015–2019: {int(ob[ob['grupo'] == 'todos']['obitos'].sum()):,}") print(resumo.round(4).to_string(index=False)) print() print(s[s["grupo"] == "esofago"][["municipio", "uf_sigla", "contexto", "observados", "rmp_uf", "obs_val", "esp_uf_val", "rmp_uf_val", "ic95_inf_uf_val", "ic95_sup_uf_val", "p_uf_val", "replica"]].round(3).to_string(index=False)) return s, resumo if __name__ == "__main__": main()