File size: 3,426 Bytes
4dbc907 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | #!/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()
|