#!/usr/bin/env python3 """ build_data.py — Gera os JSONs do dashboard Vendas Setorial a partir do BigQuery. Pipeline: 1. Le crosswalks (segmentos.json, cnae_to_segmento.csv, ncm2_to_segmento.csv). 2. Roda as queries de sql/ no BQ (cliente real, sem mock). 3. Aplica crosswalks (CNAE->segmento com override por classe; NCM2->segmento). 4. Trimestraliza RF e MDIC com shares do BCB e projeta ate 2026T1. 5. Escreve data/*.json e valida somatorios. Uso: python3 build/build_data.py Saida: data/segmentos.json, data/pagamentos.json, data/faturamento.json, data/ecommerce.json, data/combinada.json, data/meta.json Idempotente: re-rodar regenera os JSONs do zero. Nao escreve nada no BQ. """ import csv import json import logging import sys from datetime import datetime from zoneinfo import ZoneInfo from google.cloud import bigquery sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent)) import config as C # ---------- logging (TZ BR) ---------- TZ = ZoneInfo("America/Sao_Paulo") logging.Formatter.converter = lambda *a: datetime.now(TZ).timetuple() logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") log = logging.getLogger("build_data") def now_br(): return datetime.now(TZ).isoformat(timespec="seconds") # ---------- crosswalks ---------- def load_segmentos(): with open(C.XWALK_DIR / "segmentos.json", encoding="utf-8") as f: segs = json.load(f)["segmentos"] codes = [s["codigo"] for s in segs] idx = {c: i for i, c in enumerate(codes)} return segs, codes, idx def load_cnae_xwalk(): """Retorna (base_por_divisao{div2:(seg,tipo)}, override_por_classe{classe4:(seg,tipo)}).""" base, override = {}, {} with open(C.XWALK_DIR / "cnae_to_segmento.csv", encoding="utf-8") as f: for row in csv.reader(f): if not row or row[0].startswith("#") or row[0].strip() == "level": continue level, code, seg, tipo = row[0].strip(), row[1].strip(), row[2].strip(), row[3].strip() if level == "divisao": base[code] = (seg, tipo) elif level == "classe": override[code] = (seg, tipo) return base, override def load_ncm_xwalk(): """Retorna {ncm2: (seg, nome_capitulo)}.""" m = {} with open(C.XWALK_DIR / "ncm2_to_segmento.csv", encoding="utf-8") as f: for row in csv.reader(f): if not row or row[0].startswith("#") or row[0].strip() == "ncm2": continue nome = row[2].strip() if len(row) > 2 else "" m[row[0].strip().zfill(2)] = (row[1].strip(), nome) return m def cnae_digits(code): """'47.11-3' -> '4711' ; '011' -> '011' ; '47' -> '47'. So digitos.""" if not code: return "" return "".join(ch for ch in code if ch.isdigit()) def map_cnae(granularity, code, base, override): """Resolve um codigo CNAE para (segmento, tipo): classe override -> divisao base.""" d = cnae_digits(code) if granularity == "classe": c4 = d[:4] if c4 in override: return override[c4] return base.get(d[:2], ("423", "servicos")) # grupo / divisao / fallback return base.get(d[:2], ("423", "servicos")) # ---------- BQ ---------- def run_sql(client, name): sql = (C.SQL_DIR / name).read_text(encoding="utf-8") log.info("Rodando %s ...", name) rows = list(client.query(sql, location=C.BQ_LOCATION).result()) log.info(" %s -> %d linhas", name, len(rows)) return rows def q_label(tri): """'20241' -> 'T1/24'.""" return f"T{tri[4]}/{tri[2:4]}" # ---------- builders ---------- def build_pagamentos(rows, seg_idx): """Cubo compacto em colunas paralelas.""" quarters = sorted({r["trimestre"] for r in rows}) q_idx = {t: i for i, t in enumerate(quarters)} cols = {k: [] for k in ("s", "q", "f", "p", "pr", "t", "v", "n", "tw")} skipped = 0 for r in rows: if r["segmento"] not in seg_idx: skipped += 1 continue cols["s"].append(seg_idx[r["segmento"]]) cols["q"].append(q_idx[r["trimestre"]]) cols["f"].append(C.FUNCAO_IDX.get(r["funcao"], 2)) cols["p"].append(C.PAY_IDX[r["pay"]]) cols["pr"].append(C.PRES_IDX[r["pres"]]) cols["t"].append(C.TIER_IDX[r["tier"]]) cols["v"].append(round(float(r["valor"]), 2)) cols["n"].append(round(float(r["qtd"]), 2)) cols["tw"].append(round(float(r["tarifa_x_valor"]), 4)) if skipped: log.warning("pagamentos: %d linhas com segmento fora do dicionario (ignoradas)", skipped) total_v = sum(cols["v"]) return { "quarters": quarters, "quarter_labels": [q_label(t) for t in quarters], "cube": cols, "_total_valor": total_v, }, total_v def build_faturamento(rows, seg_codes, base, override): """Escolhe granularidade mais fina por ano; mapeia p/ segmento; series por regime e drill por CNAE.""" # granularidade mais fina presente por ano order = {"divisao": 0, "grupo": 1, "classe": 2} finest = {} for r in rows: a, g = r["ano"], r["granularity"] if a not in finest or order[g] > order[finest[a]]: finest[a] = g anos = sorted(finest) log.info("RF granularidade por ano: %s", {a: finest[a] for a in anos}) metrics = ["receita", "n_cnpj", "massa", "export", "arrec"] metric_col = { "receita": "receita_bruta", "n_cnpj": "n_cnpj", "massa": "massa_salarial", "export": "exportacoes", "arrec": "arrecadacao_total", } tipos = ["varejo", "atacado", "industria", "servicos"] # series[seg][tipo][regime][metric] = {ano: val} series = { s: {tp: {"normal": {m: {} for m in metrics}, "simples": {m: {} for m in metrics}} for tp in tipos} for s in seg_codes } # drill[seg][(code,tipo)] = {desc, receita, tipo} (ano mais recente, consolidado por regime) ult = max(anos) drill = {s: {} for s in seg_codes} for r in rows: a, g = r["ano"], r["granularity"] if g != finest[a]: continue seg, tp = map_cnae(g, r["cnae_code"], base, override) reg = r["regime"] for m in metrics: v = float(r[metric_col[m]] or 0) series[seg][tp][reg][m][a] = series[seg][tp][reg][m].get(a, 0.0) + v if a == ult: key = (r["cnae_code"], tp) d = drill[seg].setdefault(key, {"code": r["cnae_code"], "desc": r["cnae_desc"], "tipo": tp, "receita": 0.0}) d["receita"] += float(r["receita_bruta"] or 0) out_series = {} for s in seg_codes: out_series[s] = {} for tp in tipos: out_series[s][tp] = { reg: {m: [round(series[s][tp][reg][m].get(a, 0.0), 2) for a in anos] for m in metrics} for reg in ("normal", "simples") } out_drill = { s: sorted( [{"code": d["code"], "desc": d["desc"], "tipo": d["tipo"], "receita": round(d["receita"], 2)} for d in drill[s].values() if round(d["receita"], 2) != 0.0], key=lambda x: -x["receita"], ) for s in seg_codes } total_rb = sum( series[s][tp][reg]["receita"].get(ult, 0.0) for s in seg_codes for tp in tipos for reg in ("normal", "simples") ) return { "anos": anos, "metricas": metrics, "tipos": tipos, "series": out_series, "drill": out_drill, "_finest": {a: finest[a] for a in anos}, "_ult_ano": ult, "_total_receita_ult": total_rb, } def build_ecommerce(rows, seg_codes, ncm_map): anos = sorted({r["ano"] for r in rows}) ult = max(anos) series = {s: {"normal": {}, "simples": {}} for s in seg_codes} drill = {s: {} for s in seg_codes} # por ncm2 ncm_nome = {} for r in rows: ncm2 = str(r["ncm2"]).zfill(2) seg, nome = ncm_map.get(ncm2, ("423", "")) ncm_nome[ncm2] = nome reg = r["regime"] a = r["ano"] v = float(r["valor_bruto"] or 0) series[seg][reg][a] = series[seg][reg].get(a, 0.0) + v if a == ult: drill[seg][ncm2] = drill[seg].get(ncm2, 0.0) + v out_series = { s: {reg: [round(series[s][reg].get(a, 0.0), 2) for a in anos] for reg in ("normal", "simples")} for s in seg_codes } out_drill = { s: sorted( [{"ncm2": k, "nome": ncm_nome.get(k, ""), "valor": round(v, 2)} for k, v in drill[s].items()], key=lambda x: -x["valor"], ) for s in seg_codes } total = sum(series[s][reg].get(ult, 0.0) for s in seg_codes for reg in ("normal", "simples")) return { "anos": anos, "series": out_series, "drill": out_drill, "_ult_ano": ult, "_total_ult": total, } def build_combinada(shares_rows, fat, ecom, seg_codes): """Trimestraliza RF e MDIC com shares do BCB e projeta ate 2026T1.""" # bcb[seg][tri] = (valor_total, valor_nao_presente) bcb = {s: {} for s in seg_codes} all_tris = set() for r in shares_rows: s = r["segmento"] if s not in bcb: continue bcb[s][r["trimestre"]] = (float(r["valor_total"] or 0), float(r["valor_nao_presente"] or 0)) all_tris.add(r["trimestre"]) def quarters_of_year(y): return [f"{y}{q}" for q in (1, 2, 3, 4)] # eixo trimestral: 2008T1 ate 2026T1 tri_axis = [] for y in range(C.ANO_INICIO_COMBINADA, 2025): tri_axis += quarters_of_year(y) tri_axis += C.QUARTERS_PROJ # 20251..20254, 20261 proj_from_idx = tri_axis.index(C.QUARTERS_PROJ[0]) # eixo anual: 2008 .. 2025(proj) ano_axis = list(range(C.ANO_INICIO_COMBINADA, C.ULTIMO_ANO_ANUAL + 2)) # +1 ano projetado (2025) tipos_rf = ["todos", "varejo", "atacado", "industria", "servicos"] def rf_annual(seg, ano, tipo): """Receita RF consolidada (normal+simples) por (seg, ano, tipo). tipo='todos' soma os elos.""" if ano not in fat["anos"]: return None i = fat["anos"].index(ano) elos = fat["tipos"] if tipo == "todos" else [tipo] tot = 0.0 for tp in elos: tot += fat["series"][seg][tp]["normal"]["receita"][i] + fat["series"][seg][tp]["simples"]["receita"][i] return tot def mdic_annual(seg, ano): if ano not in ecom["anos"]: return None i = ecom["anos"].index(ano) return ecom["series"][seg]["normal"][i] + ecom["series"][seg]["simples"][i] def bcb_share(seg, ano, np=False): """Share de cada trimestre do ano (lista de 4). Fallback igual (0.25) se ano sem BCB.""" vals = [] for t in quarters_of_year(ano): tv = bcb[seg].get(t, (0.0, 0.0)) vals.append(tv[1] if np else tv[0]) tot = sum(vals) if tot <= 0: return [0.25, 0.25, 0.25, 0.25] return [v / tot for v in vals] def bcb_ratio(seg, tri, np=False): """Razao YoY do BCB: valor[tri]/valor[tri-4ano]. Fallback 1.0.""" y, q = int(tri[:4]), tri[4] cur = bcb[seg].get(tri, (0.0, 0.0)) prev = bcb[seg].get(f"{y-1}{q}", (0.0, 0.0)) cv = cur[1] if np else cur[0] pv = prev[1] if np else prev[0] if pv <= 0: return 1.0 return cv / pv def annual_from_q(qmap, ano): qs = [qmap.get(t) for t in quarters_of_year(ano)] if all(v is None for v in qs): return None return round(sum(v for v in qs if v is not None), 2) series = {} for s in seg_codes: # --- BCB trimestral (regua; realizado + real ate 2026T1) --- bcb_q = {} for t in tri_axis: tv = bcb[s].get(t) bcb_q[t] = round(tv[0], 2) if tv else None # --- RF por elo da cadeia (tipo) --- rf_q_by_tipo = {tp: {} for tp in tipos_rf} for tp in tipos_rf: rfq = rf_q_by_tipo[tp] for y in range(C.ANO_INICIO_COMBINADA, C.ULTIMO_ANO_ANUAL + 1): rf_a = rf_annual(s, y, tp) sh = bcb_share(s, y, np=False) for k, t in enumerate(quarters_of_year(y)): rfq[t] = round(rf_a * sh[k], 2) if rf_a is not None else None for t in C.QUARTERS_PROJ: y, q = int(t[:4]), t[4] prev = rfq.get(f"{y-1}{q}") rfq[t] = round(prev * bcb_ratio(s, t, np=False), 2) if prev is not None else None # --- MDIC (e-commerce = varejo online; sem dimensao de cadeia) --- mdic_q = {} for y in range(C.ANO_INICIO_COMBINADA, C.ULTIMO_ANO_ANUAL + 1): mdic_a = mdic_annual(s, y) shn = bcb_share(s, y, np=True) for k, t in enumerate(quarters_of_year(y)): mdic_q[t] = round(mdic_a * shn[k], 2) if mdic_a is not None else None for t in C.QUARTERS_PROJ: y, q = int(t[:4]), t[4] prev = mdic_q.get(f"{y-1}{q}") mdic_q[t] = round(prev * bcb_ratio(s, t, np=True), 2) if prev is not None else None tri_series = { "quarters": tri_axis, "quarter_labels": [q_label(t) for t in tri_axis], "proj_from_idx": proj_from_idx, "bcb": [bcb_q.get(t) for t in tri_axis], "mdic": [mdic_q.get(t) for t in tri_axis], "rf": {tp: [rf_q_by_tipo[tp].get(t) for t in tri_axis] for tp in tipos_rf}, } bcb_an = [annual_from_q(bcb_q, y) for y in ano_axis] mdic_an = [] rf_an = {tp: [] for tp in tipos_rf} for y in ano_axis: if y <= C.ULTIMO_ANO_ANUAL: mdic_an.append(mdic_annual(s, y)) for tp in tipos_rf: rf_an[tp].append(rf_annual(s, y, tp)) else: # ano projetado = soma dos trimestres projetados mdic_an.append(annual_from_q(mdic_q, y)) for tp in tipos_rf: rf_an[tp].append(annual_from_q(rf_q_by_tipo[tp], y)) an_series = { "anos": ano_axis, "proj_from_idx": ano_axis.index(C.ULTIMO_ANO_ANUAL + 1), "bcb": bcb_an, "mdic": mdic_an, "rf": rf_an, } series[s] = {"trimestral": tri_series, "anual": an_series} return {"series": series, "_proj_quarters": C.QUARTERS_PROJ, "tipos_rf": tipos_rf} def build_mercado(rows5, fat, ecom, seg_codes): """Penetracao de cartao por segmento (anual, sem projecao). 'total': todas as capturas / faturamento das empresas (RF). 'ecommerce': cartao nao presente / faturamento e-commerce (MDIC). Componentes do numerador permitem os drills credito/debito, tier, a vista x parcelado e nº parcelas.""" # Denominador da aba "mercado" = faturamento de VAREJO (inclui atacarejo/supermercados, ja tipo=varejo). # Janela restrita aos anos com granularidade CNAE 'classe' (onde o filtro fino de varejo existe); # antes disso o varejo fica agregado em segmentos genericos e a razao quebra (ex.: vestuario 2015->2016). finest = fat.get("_finest", {}) anos_total = [a for a in fat["anos"] if finest.get(a) == "classe"] if not anos_total: # fallback defensivo anos_total = fat["anos"] anos_ecom = ecom["anos"] PARC = {"1 parcela": "p1", "2 e 3 parcelas": "p23", "4 a 6 parcelas": "p46", "7 ou + parcelas": "p7"} comp = ["card", "credito", "debito_outros", "cred_prem_emp", "cred_avista", "cred_parcelado", "p1", "p23", "p46", "p7"] def blank(anos): return {k: {a: 0.0 for a in anos} for k in comp} acc = {"total": {s: blank(anos_total) for s in seg_codes}, "ecommerce": {s: blank(anos_ecom) for s in seg_codes}} for r in rows5: seg = r["segmento"] if seg not in acc["total"]: continue ano, funcao, pres, tier, parc = r["ano"], r["funcao"], r["pres"], r["tier"], r["parcelas"] v = float(r["valor"] or 0) scopes = [] if ano in anos_total: scopes.append("total") # todas as capturas if pres == "nao_presente" and ano in anos_ecom: scopes.append("ecommerce") # apenas cartao nao presente for sc in scopes: a = acc[sc][seg] a["card"][ano] += v if funcao == "Crédito": a["credito"][ano] += v if tier in ("premium", "corp"): a["cred_prem_emp"][ano] += v if parc == "1 parcela": a["cred_avista"][ano] += v else: a["cred_parcelado"][ano] += v pk = PARC.get(parc) if pk: a[pk][ano] += v elif funcao in ("Débito", "Pré-pago"): a["debito_outros"][ano] += v def rf_varejo(seg, ano): """Denominador da aba mercado = faturamento de VAREJO apenas (consolidado normal+simples).""" i = fat["anos"].index(ano) return sum(fat["series"][seg]["varejo"][rg]["receita"][i] for rg in ("normal", "simples")) def mdic_total(seg, ano): i = ecom["anos"].index(ano) return ecom["series"][seg]["normal"][i] + ecom["series"][seg]["simples"][i] def emit(scope, seg, anos, fat_fn): a = acc[scope][seg] block = {"fat": [round(fat_fn(seg, an), 2) for an in anos]} for k in comp: block[k] = [round(a[k][an], 2) for an in anos] return block out = {"anos_total": anos_total, "anos_ecom": anos_ecom, "total": {}, "ecommerce": {}} for s in seg_codes: out["total"][s] = emit("total", s, anos_total, rf_varejo) out["ecommerce"][s] = emit("ecommerce", s, anos_ecom, mdic_total) return out # ---------- validacao ---------- def validate(pag, fat, ecom, comb, merc, seg_codes): log.info("=== Validacao ===") ok = True # combinada: soma trimestres realizados == anual realizado (tolerancia), tipo='todos' sample = seg_codes[15] # '416' Roupas s = comb["series"][sample]["anual"] i2024 = s["anos"].index(2024) rf_anual_2024 = s["rf"]["todos"][i2024] tri = comb["series"][sample]["trimestral"] rf_tri_2024 = sum(v for t, v in zip(tri["quarters"], tri["rf"]["todos"]) if t.startswith("2024") and v is not None) if rf_anual_2024 and abs(rf_tri_2024 - rf_anual_2024) / rf_anual_2024 > 0.001: log.error(" FAIL: soma trimestral RF 2024 (%s) != anual (%s) p/ %s", rf_tri_2024, rf_anual_2024, sample) ok = False else: log.info(" OK: trimestral RF 2024 == anual (seg %s)", sample) # projecao chega a 2026T1 if comb["series"][sample]["trimestral"]["quarters"][-1] != "20261": log.error(" FAIL: serie trimestral nao termina em 20261") ok = False else: log.info(" OK: projecao chega a 2026T1") def rb_seg(sg): return sum( sum(fat["series"][sg][tp][rg]["receita"]) for tp in fat["tipos"] for rg in ("normal", "simples") ) rb423 = rb_seg("423") rb_all = sum(rb_seg(sg) for sg in seg_codes) # log da distribuicao por tipo (sanidade do novo eixo de cadeia) ult_i = len(fat["anos"]) - 1 by_tipo = {tp: 0.0 for tp in fat["tipos"]} for sg in seg_codes: for tp in fat["tipos"]: by_tipo[tp] += fat["series"][sg][tp]["normal"]["receita"][ult_i] + fat["series"][sg][tp]["simples"]["receita"][ult_i] tot_tipo = sum(by_tipo.values()) or 1 log.info(" Cadeia (ult ano): " + " · ".join(f"{tp}={by_tipo[tp]/tot_tipo*100:.0f}%" for tp in fat["tipos"])) share423 = rb423 / rb_all if rb_all else 0 log.info(" Cobertura: segmento 'Outros'(423) = %.1f%% da receita total RF", share423 * 100) if share423 > 0.35: log.warning(" ATENCAO: 423 concentra >35%% da receita; revisar crosswalk") log.info(" pagamentos: %d combos no cubo, %d trimestres", len(pag["cube"]["v"]), len(pag["quarters"])) log.info(" ecommerce: %d anos, total ult ano R$ %.1f bi", len(ecom["anos"]), ecom["_total_ult"] / 1e9) # mercado (penetracao de cartao): identidades + sanidade smp = "416" b = merc["total"][smp] j = len(merc["anos_total"]) - 1 id1 = abs((b["credito"][j] + b["debito_outros"][j]) - b["card"][j]) id2 = abs((b["cred_avista"][j] + b["cred_parcelado"][j]) - b["credito"][j]) id3 = abs((b["p1"][j] + b["p23"][j] + b["p46"][j] + b["p7"][j]) - b["credito"][j]) if max(id1, id2, id3) > 1.0: log.error(" FAIL: identidades de cartao nao fecham (id1=%.2f id2=%.2f id3=%.2f)", id1, id2, id3) ok = False else: log.info(" OK: identidades cartao fecham (credito+debito=card; avista+parcelado=credito; parcelas=credito)") pen = b["card"][j] / b["fat"][j] if b["fat"][j] else 0 log.info(" Mercado: penetracao cartao seg 416 (%d) = %.1f%% | e-commerce %d anos", merc["anos_total"][j], pen * 100, len(merc["anos_ecom"])) return ok def strip_private(d): return {k: v for k, v in d.items() if not k.startswith("_")} def main(): log.info("Build iniciado %s", now_br()) segs, seg_codes, seg_idx = load_segmentos() base, override = load_cnae_xwalk() ncm_map = load_ncm_xwalk() log.info("Crosswalks: %d segmentos, %d divisoes base, %d overrides classe, %d capitulos NCM", len(seg_codes), len(base), len(override), len(ncm_map)) client = bigquery.Client(project=C.BQ_PROJECT, location=C.BQ_LOCATION) pag_rows = run_sql(client, "v1_pagamentos_cube.sql") fat_rows = run_sql(client, "v2_faturamento.sql") ecom_rows = run_sql(client, "v3_ecommerce.sql") shares_rows = run_sql(client, "v4_bcb_shares.sql") cartao_rows = run_sql(client, "v5_cartao_anual.sql") pag, _ = build_pagamentos(pag_rows, seg_idx) fat = build_faturamento(fat_rows, seg_codes, base, override) ecom = build_ecommerce(ecom_rows, seg_codes, ncm_map) comb = build_combinada(shares_rows, fat, ecom, seg_codes) merc = build_mercado(cartao_rows, fat, ecom, seg_codes) ok = validate(pag, fat, ecom, comb, merc, seg_codes) C.DATA_DIR.mkdir(exist_ok=True) meta = { "gerado_em": now_br(), "rotulos": {"pagamentos": C.ROTULO_PAGAMENTOS, "faturamento": C.ROTULO_FATURAMENTO, "ecommerce": C.ROTULO_ECOMMERCE, "fonte": C.FONTE_PUBLICA}, "proj_quarters": [q_label(t) for t in C.QUARTERS_PROJ], } out = { "segmentos.json": {"segmentos": segs}, "pagamentos.json": strip_private(pag), "faturamento.json": strip_private(fat), "ecommerce.json": strip_private(ecom), "combinada.json": comb, "mercado.json": merc, "meta.json": meta, } for fname, payload in out.items(): p = C.DATA_DIR / fname p.write_text(json.dumps(payload, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") log.info("Escrito %s (%.1f KB)", fname, p.stat().st_size / 1024) log.info("Build concluido %s | validacao=%s", now_br(), "OK" if ok else "FALHOU") sys.exit(0 if ok else 2) if __name__ == "__main__": main()