Blame for scripts/52_telemetria_v050.py

#!/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()