v0.7.0: FNH gate de punição + árvore de pesos indexada (arestas duplas) + quebra-cabeça do Pensamento + NLP/NLG + decodificador especulativo + auto-ajuste SOM limitado + correções v070-1/2/3 — 73/73 gates, 84/84 cargo, provas r-28..r-32
4dbc907 verified | #!/usr/bin/env python3 | |
| """65_telemetria_v070.py — v0.7.0: PAINEL CONSOLIDADO de telemetria de todo | |
| o projeto (requisito: "faça telemetria em todo o projeto"). | |
| Agrega: | |
| - telemetria_fase_c.jsonl (treino conjunto: CE, confiança, conflito); | |
| - treino_v070.json (fases A/B/C + árvore + FNH); | |
| - arvore_quebra_v070.json (quebra-cabeça real); | |
| - suite_fnh/nlp_nlg (gates dos módulos novos); | |
| - varredura de TODAS as séries com detecção de aprendizado (Theil-Sen | |
| polaridade-aware — herdado v0.6.0) aplicada às séries da fase C. | |
| """ | |
| import json, os, sys, math | |
| def teilen_sen(xs, ys): | |
| """Inclinação Theil-Sen (mediana dos pares) — robusta a outliers.""" | |
| incl = [] | |
| for i in range(len(xs)): | |
| for j in range(i + 1, len(xs)): | |
| if xs[j] != xs[i]: | |
| incl.append((ys[j] - ys[i]) / (xs[j] - xs[i])) | |
| if not incl: | |
| return 0.0 | |
| incl.sort() | |
| return incl[len(incl) // 2] | |
| def main(): | |
| pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| R = os.path.join(pasta, "resultados") | |
| painel = {"versao": "telemetria-0.7.0", "modulos": {}} | |
| # ---- série da fase C: CE + confiança ao longo dos passos | |
| tel_c = os.path.join(R, "telemetria_fase_c.jsonl") | |
| series = {} | |
| if os.path.exists(tel_c): | |
| with open(tel_c) as f: | |
| for linha in f: | |
| try: | |
| ev = json.loads(linha) | |
| except Exception: | |
| continue | |
| if ev.get("tipo") != "dump": | |
| continue | |
| dados = ev.get("dados", {}) | |
| for k, v in dados.items(): | |
| # séries no formato [[passo, valor], ...] | |
| if isinstance(v, list) and v and isinstance(v[0], list): | |
| series[k] = [par[1] for par in v if len(par) == 2] | |
| deteccoes = {} | |
| for k, vs in series.items(): | |
| if len(vs) < 8: | |
| continue | |
| xs = list(range(len(vs))) | |
| incl = teilen_sen(xs, vs) | |
| deteccoes[k] = { | |
| "inclinacao_theil_sen": incl, | |
| "n": len(vs), | |
| "primeiro": vs[0], "ultimo": vs[-1], | |
| } | |
| painel["fase_c_series"] = deteccoes | |
| # ---- módulos | |
| for nome, arq in [("treino", "treino_v070.json"), | |
| ("fnh", "suite_fnh_v070.json"), | |
| ("arvore_quebra", "arvore_quebra_v070.json"), | |
| ("nlp_nlg", "suite_nlp_nlg_v070.json")]: | |
| p = os.path.join(R, arq) | |
| if os.path.exists(p): | |
| with open(p) as f: | |
| painel["modulos"][nome] = json.load(f) | |
| # ---- resumo de direções (polaridade: CE/perda ↓ = aprendendo) | |
| aprendendo = 0 | |
| piorando = 0 | |
| for k, d in deteccoes.items(): | |
| klow = k.lower() | |
| pol = -1 if ("perda" in klow or "ce" in klow or "erro" in klow) else +1 | |
| prod = d["inclinacao_theil_sen"] * pol | |
| if prod < -1e-12: | |
| aprendendo += 1 | |
| elif prod > 1e-12: | |
| piorando += 1 | |
| painel["resumo_direcoes"] = {"aprendendo": aprendendo, "piorando": piorando} | |
| saida = os.path.join(R, "telemetria_v070.json") | |
| with open(saida, "w") as f: | |
| json.dump(painel, f, indent=1, ensure_ascii=False) | |
| print(f"séries analisadas: {len(deteccoes)} | " | |
| f"aprendendo {aprendendo} / piorando {piorando}") | |
| print(f"OK: {saida}") | |
| if __name__ == "__main__": | |
| main() | |