v0.5.0: suíte SOM-4D + suíte JEPA (12) + tecnologia Pensamento + telemetria Theil–Sen — 29/29 gates verdes (provas r-18..r-22)
e26f4cc verified | #!/usr/bin/env python3 | |
| """52_telemetria_v050.py — Telemetria: detecção de aprendizado, ajustes | |
| automatizados e diagnóstico de erros (typos primeiro). | |
| Demonstra (r-22 §telemetria): | |
| 1. Detecção robusta de aprendizado (Theil–Sen + R²) — incluindo o caso com | |
| OUTLIERS que enganaria a média OLS (Teo TD-1); | |
| 2. Diagnóstico de erros comuns de refatoração: typo em chave de config → | |
| sugestão por Levenshtein (lint_config); classificação de erro; | |
| 3. Painel de telemetria: séries, vereditos, escore de aprendizado. | |
| Saída: resultados/telemetria_v050.json | |
| """ | |
| import json, os | |
| def main(): | |
| import khtst_rust.nucleo_rust as nr | |
| pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| res = {"versao": "0.5.0", "demos": {}} | |
| # ------------------------------------------------ 1. Theil–Sen robusto | |
| print("=== Detecção de aprendizado (Theil–Sen robusto) ===", flush=True) | |
| # série limpa: perda decaindo com ruído leve | |
| import random | |
| rng = random.Random(5) | |
| limpa = [2.0 * (0.97 ** t) + rng.gauss(0, 0.02) for t in range(60)] | |
| # série com 20% de outliers violentos (OOM momentâneo, spike de NaN-clip) | |
| suja = list(limpa) | |
| for i in range(0, 60, 5): | |
| suja[i] = 50.0 + rng.gauss(0, 5) | |
| # implementação local do detector (o painel completo roda dentro do Rust | |
| # durante os treinos; aqui verificamos os mesmos critérios) | |
| def theil_sen(ys): | |
| incl = [] | |
| for i in range(len(ys)): | |
| for j in range(i + 1, len(ys)): | |
| if j > i: | |
| incl.append((ys[j] - ys[i]) / (j - i)) | |
| incl.sort() | |
| return incl[len(incl) // 2] | |
| incl_limpa = theil_sen(limpa) | |
| incl_suja = theil_sen(suja) | |
| ols_limpa = sum(limpa[i + 1] - limpa[i] for i in range(len(limpa) - 1)) / (len(limpa) - 1) | |
| ols_suja = sum(suja[i + 1] - suja[i] for i in range(len(suja) - 1)) / (len(suja) - 1) | |
| print(f" limpa: Theil-Sen {incl_limpa:+.5f} | OLS {ols_limpa:+.5f} (concordam)") | |
| print(f" com 20% outliers: Theil-Sen {incl_suja:+.5f} | OLS {ols_suja:+.5f} (OLS enganado)") | |
| # Teo TD-1: Theil-Sen permanece ACURADO com 20% de outliers (desvio < 50% | |
| # do valor verdadeiro) enquanto o OLS é enganado (desvio > 10x) | |
| ok_ts = ( | |
| incl_limpa < 0 | |
| and abs(incl_suja - incl_limpa) < 0.5 * abs(incl_limpa) | |
| and abs(ols_suja - incl_limpa) > 10.0 * abs(incl_limpa) | |
| ) | |
| res["demos"]["deteccao_robusta"] = { | |
| "inclinacao_theil_sen_limpa": incl_limpa, | |
| "inclinacao_ols_limpa": ols_limpa, | |
| "inclinacao_theil_sen_com_outliers": incl_suja, | |
| "inclinacao_ols_com_outliers": ols_suja, | |
| "gate_td1": ok_ts, | |
| } | |
| # ------------------------------------------------ 2. diagnóstico de typos | |
| print("\n=== Diagnóstico de erros (typo em config → sugestão) ===", flush=True) | |
| cfg_ruim = os.path.join(pasta, "resultados", "_cfg_typo_tmp.json") | |
| os.makedirs(os.path.dirname(cfg_ruim), exist_ok=True) | |
| with open(cfg_ruim, "w") as f: | |
| json.dump({"treino": {"passos_fase_a": 10}, "ncm": {"nivei": [4, 8], "k_por_nivl": 64}}, f) | |
| avisos = nr.lint_config(cfg_ruim) | |
| os.remove(cfg_ruim) | |
| sugestoes = {} | |
| for a in avisos: | |
| sugestoes[a["chave"]] = a.get("sugestao") | |
| print(f" {a['chave']} -> sugerido: {a.get('sugestao')} ({a.get('tipo')})") | |
| ok_typo = ("nivei" in str(sugestoes) or any("niveis" == s for s in sugestoes.values() if s)) | |
| res["demos"]["diagnostico_typo"] = { | |
| "avisos": [{k: getattr(a, k, None) if not isinstance(a, dict) else a.get(k) for k in (("chave", "sugestao", "tipo"))} for a in (avisos if isinstance(avisos, list) else [])], | |
| "gate_typo_detectado": ok_typo, | |
| } | |
| # classificação de erro por taxonomia (erro comum de refatoração) | |
| # dois exemplos realistas: índice fora (typo em config) e NaN (lr alto) | |
| classe = nr.classificar_erro("index out of bounds: id 64 >= K_por_nivel") | |
| print(f" classificar_erro('shape mismatch...') -> {classe[:2]}") | |
| res["demos"]["classificacao_erro"] = { | |
| "exemplo": [[c[0], c[1]] for c in classe], | |
| "gate_classifica": len(classe) > 0, | |
| } | |
| res["demos"]["cobertura_api"] = { | |
| "funcoes_v050": [f for f in dir(nr) if not f.startswith("_")], | |
| } | |
| # ------------------------------------------------------------------ resumo | |
| gates = [] | |
| for n, s in res["demos"].items(): | |
| if isinstance(s, dict): | |
| gates += [(f"{n}.{g[5:]}", v) for g, v in s.items() if g.startswith("gate_")] | |
| ok = sum(1 for _, v in gates if v) | |
| res["resumo"] = {"gates_totais": len(gates), "gates_ok": ok, "todos_verdes": ok == len(gates)} | |
| print(f"\n=== RESUMO: {ok}/{len(gates)} gates verdes ===") | |
| with open(os.path.join(pasta, "resultados", "telemetria_v050.json"), "w") as f: | |
| json.dump(res, f, indent=1, ensure_ascii=False) | |
| print("salvo: resultados/telemetria_v050.json") | |
| if not res["resumo"]["todos_verdes"]: | |
| raise SystemExit(1) | |
| if __name__ == "__main__": | |
| main() | |