"""som_auto_adjust.py — V6.5-V2-auto-adjust: Auto-ajuste dos Mapas Auto-Organizáveis (Kohonen SOM) com base nas 4 métricas principais e 4 indicadores de falha. ============================================================================ USER REQUIREMENT ============================================================================ "ANALISAR (ATENÇÃO: CORRIGIR e APRIMORAR matematicamente os erros de NaN que estejam ocorrendo por explosão de gradientes dos pesos) matematicamente e logicamente e inserir melhorias para os scripts das métricas de aprendizado e indicadores de falha em Mapas Auto-Organizáveis (SOM / Redes de Kohonen) avaliam a fidelidade de representação dos dados e a preservação da vizinhança topológica" + "aprimorar matematicamente a lógica de autoajustes analisando Kohonen" + "PORTANTO: observar se os valores demonstram que o modelo esteja aprendendo e parar caso não esteja". ============================================================================ AUTO-AJUSTES IMPLEMENTADOS (1 por métrica/indicador) ============================================================================ 1.1 QE alta → +α₀_eff (acelera aprendizado momentâneo), revival agressivo 1.2 TE alta → +σ₀_eff (expande vizinhança para corrigir topologia) 1.3 KL alto → combinado: +α e +σ (intervenção conjunta) 1.4 VE baixa → revive neurônios mortos (representação insuficiente) 2.1 Topological Collapse → reinit pesos SOM (data-driven, N(0, 0.1) fallback) 2.2 Dead Neurons > 50% → revive_dead_neurons() + +conscience_γ 2.3 QE Stagnation → +α₀_eff (sobe aprendizado), bypass decaimento 2.4 Neighborhood Crossing → -α₀_eff (desacelera para estabilizar), +σ₀_eff Stop-learning detection (user: "parar caso não esteja"): - Collapse total detectado (severity="point") E revival falhou - QE estagnado em nível crítico (severity="severe") por ≥ 5 chunks consecutivos - Dead neuron rate > 0.8 após revival + conscience boost - All metrics NaN por 2 batches consecutivos - V7: Neighborhood crossing severity="severe" por ≥ 4 chunks consecutivos (instabilidade sináptica crônica — Prova 12) ============================================================================ MATH (Kohonen canonical + adaptive extensions) ============================================================================ - α_t = α₀ · exp(-t / τ_α), τ_α = 2000, α_min = 0.001 (FIX-3 floor) - σ_t = σ₀ · exp(-t / τ_σ), τ_σ = 1000, σ_min = 0.1 (FIX-3 floor) - Auto-boost (multiplicative sobre α₀_eff e σ₀_eff): α₀_eff ← α₀_eff · (1 + 0.3 · norm_qe) (resposta ao QE) σ₀_eff ← σ₀_eff · (1 + 0.4 · te) (resposta ao TE) α₀_eff ← α₀_eff · (1 - 0.2 · crossing_severity) (estabiliza crossing) - V7 — Crossing severo extra boost: σ₀_eff ← σ₀_eff + 0.25 (re-organização topológica forçada) - Decay bound: α₀_eff ∈ [α₀, 2·α₀], σ₀_eff ∈ [σ₀, 2·σ₀] (nunca abaixo do canônico, no máximo 2× para evitar divergência) Prova 12 (neighborhood_crossing_severe stop): Se std(TE_final) > 2·growth_threshold·std(TE_initial) ou TE cresce monotonicamente no final, severity="severe". Após 4 chunks consecutivos severe, o treino para — a rede entrou em regime de instabilidade sináptica crônica onde correções (-α +σ) não recuperam a topologia. Continuar o treino só degrada mais a topologia, desperdiçando compute. Stop-learning preserva o estado atual e permite diagnóstico. """ from __future__ import annotations import logging import math from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple import torch logger = logging.getLogger(__name__) # ============================================================================ # 1. SOMAutoAdjustState — Estado persistente entre batches # ============================================================================ @dataclass class SOMAutoAdjustState: """Estado persistente do auto-ajuste entre batches. Mantém contadores de anomalias consecutivas para decidir quando parar o treino (user requirement: "parar caso não esteja"). """ # Boosts efetivos (multiplicativos sobre α₀ e σ₀ canônicos) alpha0_boost: float = 1.0 # multiplicador sobre α₀ sigma0_boost: float = 1.0 # multiplicador sobre σ₀ # Contadores de anomalias consecutivas (para stop-learning) consecutive_collapse_point: int = 0 consecutive_qe_stagnant_severe: int = 0 consecutive_dead_rate_critical: int = 0 consecutive_nan_batches: int = 0 # V7 — Neighborhood crossing severo consecutivo (instabilidade crônica) # User requirement: "neighborhood_crossing_severe" deve ser detectado e # tratado. Após N batches consecutivos com severity="severe", o treino # para (a rede sofreu instabilidade sináptica crônica). consecutive_crossing_severe: int = 0 # Histórico de ações tomadas (últimas 50) actions_log: List[Dict[str, Any]] = field(default_factory=list) # Flag de parada (uma vez True, treino não deve continuar) stop_training: bool = False stop_reason: Optional[str] = None # Total de ações aplicadas n_actions_total: int = 0 def reset_consecutive_counters(self) -> None: """Reseta contadores consecutivos (chamado após batch saudável).""" self.consecutive_collapse_point = 0 self.consecutive_qe_stagnant_severe = 0 self.consecutive_dead_rate_critical = 0 self.consecutive_nan_batches = 0 self.consecutive_crossing_severe = 0 def to_dict(self) -> Dict[str, Any]: return { "alpha0_boost": float(self.alpha0_boost), "sigma0_boost": float(self.sigma0_boost), "consecutive_collapse_point": int(self.consecutive_collapse_point), "consecutive_qe_stagnant_severe": int(self.consecutive_qe_stagnant_severe), "consecutive_dead_rate_critical": int(self.consecutive_dead_rate_critical), "consecutive_nan_batches": int(self.consecutive_nan_batches), "consecutive_crossing_severe": int(self.consecutive_crossing_severe), "stop_training": bool(self.stop_training), "stop_reason": self.stop_reason, "n_actions_total": int(self.n_actions_total), "n_actions_logged": int(len(self.actions_log)), "recent_actions": self.actions_log[-10:], } # ============================================================================ # 2. SOMAutoAdjuster — Aplica correções com base nas métricas # ============================================================================ class SOMAutoAdjuster: """Aplica auto-ajustes ao SOM com base nas 8 métricas canônicas. Uso: adjuster = SOMAutoAdjuster() # A cada batch: action = adjuster.adjust(kls, som_metrics_dict) if action["stop_training"]: break # parar caso não esteja aprendendo """ # Thresholds (Kohonen canonical + heuristic) QE_HIGH_THRESHOLD = 0.5 # QE > 0.5 indica baixa resolução TE_HIGH_THRESHOLD = 0.3 # TE > 0.3 indica topologia fraca KL_HIGH_THRESHOLD = 0.4 # KL > 0.4 indica problema conjunto VE_LOW_THRESHOLD = 0.3 # VE < 0.3 indica representação fraca DEAD_RATE_REVIVE_THRESHOLD = 0.5 # revive se > 50% mortos DEAD_RATE_CRITICAL_THRESHOLD = 0.8 # crítico se > 80% mortos QE_STAGNATION_SEVERE_LIMIT = 5 # 5 batches consecutivos = parar COLLAPSE_POINT_LIMIT = 3 # 3 collapses consecutivos = parar DEAD_CRITICAL_LIMIT = 5 # 5 batches críticos = parar NAN_BATCH_LIMIT = 2 # 2 batches NaN = parar # V7 — Neighborhood crossing severo (user requirement) # 4 batches consecutivos com crossing_severity="severe" = parar treino. # Justificativa matemática (Prova 12): se a oscilação crônica do TE # persiste por 4+ batches apesar das correções (-α +σ), isto indica # que a vizinhança topológica não consegue fixar — a rede entrou em # regime de instabilidade sináptica crônica, e continuar o treino # só degrada a topologia. Stop-learning evita desperdiçar compute. CROSSING_SEVERE_LIMIT = 4 def __init__(self, state: Optional[SOMAutoAdjustState] = None): self.state = state or SOMAutoAdjustState() # ------------------------------------------------------------------ # API principal: adjust() # ------------------------------------------------------------------ def adjust( self, kls: Any, # KohonenLearningSystemV2 som_metrics: Dict[str, Any], batch_idx: int = 0, ) -> Dict[str, Any]: """Aplica auto-ajustes ao SOM com base nas métricas. Modifica kls.som (alpha0_eff, sigma0_eff, conscience_gamma, pesos) e registra ações no estado. Retorna dict com ações aplicadas e flag stop_training (user: "parar caso não esteja aprendendo"). Args: kls: instância do KohonenLearningSystemV2. som_metrics: dict retornado por compute_all_metrics(). batch_idx: índice do batch (para log). Returns: Dict com: - actions: list[str] — ações tomadas - stop_training: bool — se True, parar o treino - stop_reason: str | None - state: dict — estado atual do adjuster - boosts: dict — alpha0_boost, sigma0_boost atuais """ if som_metrics is None: return self._no_action_response(batch_idx, reason="no_metrics") actions: List[str] = [] boosts_before = (self.state.alpha0_boost, self.state.sigma0_boost) # Detection helpers nan_detected = bool(som_metrics.get("nan_detected", False)) qe = float(som_metrics.get("quantization_error", 0.0)) te = float(som_metrics.get("topological_error", 0.0)) kl = float(som_metrics.get("kaski_lagus_error", 0.0)) ve = float(som_metrics.get("explained_variance_share", 0.0)) collapse = som_metrics.get("topological_collapse", {}) or {} collapse_severity = str(collapse.get("severity", "none")) collapsed_to_point = bool(collapse.get("collapsed_to_point", False)) dead = som_metrics.get("dead_neuron_rate", {}) or {} dead_rate = float(dead.get("dead_neuron_rate", 0.0)) n_dead = int(dead.get("n_dead", 0)) n_total = int(dead.get("n_total", 0)) qe_stag = som_metrics.get("qe_stagnation", {}) or {} qe_stagnant = bool(qe_stag.get("is_stagnant", False)) qe_stag_severity = str(qe_stag.get("severity", "none")) nb_cross = som_metrics.get("neighborhood_crossing", {}) or {} crossing_detected = bool(nb_cross.get("detected", False)) crossing_severity = str(nb_cross.get("severity", "none")) # ============================================================== # NaN batch (qualquer anomalia detectada como nan) # ============================================================== if nan_detected: self.state.consecutive_nan_batches += 1 actions.append(f"nan_detected (consecutive={self.state.consecutive_nan_batches})") if self.state.consecutive_nan_batches >= self.NAN_BATCH_LIMIT: self.state.stop_training = True self.state.stop_reason = ( f"nan_batches_consecutive={self.state.consecutive_nan_batches}" ) logger.error( f"[som_auto_adjust] STOP: {self.state.stop_reason} " f"(NaN propagado por {self.state.consecutive_nan_batches} batches)" ) else: self.state.consecutive_nan_batches = 0 # ============================================================== # 2.1 Topological Collapse — reinit pesos (data-driven) # ============================================================== if collapsed_to_point: self.state.consecutive_collapse_point += 1 actions.append(f"collapse_point_detected (consecutive={self.state.consecutive_collapse_point})") # Tenta revival agressivo (data-driven) try: revival = kls.som.revive_dead_neurons( data_buffer=list(kls.buffer_4d), dead_threshold=0, # revive TODOS mortos ) actions.append( f"revive_after_collapse: n_revived={revival.get('n_revived', 0)}" ) # Boost agressivo de conscience self.state.alpha0_boost = min(2.0, self.state.alpha0_boost * 1.3) self.state.sigma0_boost = min(2.0, self.state.sigma0_boost * 1.5) actions.append("boost_after_collapse: α×1.3, σ×1.5") except Exception as e: actions.append(f"revive_after_collapse_failed: {str(e)[:100]}") if self.state.consecutive_collapse_point >= self.COLLAPSE_POINT_LIMIT: self.state.stop_training = True self.state.stop_reason = ( f"collapse_point_consecutive={self.state.consecutive_collapse_point}" ) logger.error( f"[som_auto_adjust] STOP: {self.state.stop_reason}" ) else: self.state.consecutive_collapse_point = 0 # ============================================================== # 1.1 QE alta → +α₀ (acelera aprendizado) # ============================================================== if qe > self.QE_HIGH_THRESHOLD and math.isfinite(qe): boost = 0.3 * min(qe, 2.0) # boost proporcional ao QE, capped self.state.alpha0_boost = min(2.0, self.state.alpha0_boost + boost) actions.append(f"qe_high_boost: α+{boost:.3f} (qe={qe:.3f})") # ============================================================== # 1.2 TE alta → +σ₀ (expande vizinhança) # ============================================================== if te > self.TE_HIGH_THRESHOLD and math.isfinite(te): boost = 0.4 * min(te, 1.0) self.state.sigma0_boost = min(2.0, self.state.sigma0_boost + boost) actions.append(f"te_high_boost: σ+{boost:.3f} (te={te:.3f})") # ============================================================== # 1.3 KL alto → combinado (já coberto por QE+TE acima) # ============================================================== if kl > self.KL_HIGH_THRESHOLD and math.isfinite(kl): # Log only — QE e TE já foram tratados individualmente actions.append(f"kl_high: kl={kl:.3f} (covered by qe/te boosts)") # ============================================================== # 1.4 VE baixa → revival adicional (representação insuficiente) # ============================================================== if ve < self.VE_LOW_THRESHOLD and math.isfinite(ve): # Força revival de neurônios mortos para melhorar cobertura try: if dead_rate > 0.3: revival = kls.som.revive_dead_neurons( data_buffer=list(kls.buffer_4d), dead_threshold=0, ) actions.append( f"ve_low_revive: n_revived={revival.get('n_revived', 0)} (ve={ve:.3f})" ) except Exception as e: actions.append(f"ve_low_revive_failed: {str(e)[:100]}") # ============================================================== # 2.2 Dead Neurons > 50% → revive + +conscience_γ # ============================================================== if dead_rate > self.DEAD_RATE_REVIVE_THRESHOLD: try: revival = kls.auto_revive_if_needed( dead_rate_threshold=self.DEAD_RATE_REVIVE_THRESHOLD, min_steps_between_revivals=50, # permite mais frequentes ) if revival.get("action") == "auto_revived": actions.append( f"dead_rate_revive: n_revived={revival.get('n_revived', 0)}, " f"dead_rate {revival.get('dead_rate_before', 0):.3f} → " f"{revival.get('dead_rate_after', 0):.3f}" ) # Aumenta conscience_gamma para distribuir BMU melhor if hasattr(kls.som, 'auto_adjust_conscience_gamma'): gamma_result = kls.som.auto_adjust_conscience_gamma() actions.append( f"conscience_auto_adjust: γ {gamma_result['gamma_old']:.3f} → " f"{gamma_result['gamma_new']:.3f} (r={gamma_result['activation_rate']:.3f})" ) except Exception as e: actions.append(f"dead_rate_revive_failed: {str(e)[:100]}") if dead_rate > self.DEAD_RATE_CRITICAL_THRESHOLD: self.state.consecutive_dead_rate_critical += 1 if self.state.consecutive_dead_rate_critical >= self.DEAD_CRITICAL_LIMIT: self.state.stop_training = True self.state.stop_reason = ( f"dead_rate_critical_consecutive=" f"{self.state.consecutive_dead_rate_critical}" ) logger.error( f"[som_auto_adjust] STOP: {self.state.stop_reason}" ) else: self.state.consecutive_dead_rate_critical = 0 else: self.state.consecutive_dead_rate_critical = 0 # ============================================================== # 2.3 QE Stagnation → +α₀ (sobe aprendizado, bypass decaimento) # ============================================================== if qe_stagnant: if qe_stag_severity == "severe": self.state.consecutive_qe_stagnant_severe += 1 # Boost agressivo self.state.alpha0_boost = min(2.0, self.state.alpha0_boost + 0.2) actions.append( f"qe_stagnant_severe_boost: α+0.2 (consecutive=" f"{self.state.consecutive_qe_stagnant_severe})" ) if self.state.consecutive_qe_stagnant_severe >= self.QE_STAGNATION_SEVERE_LIMIT: self.state.stop_training = True self.state.stop_reason = ( f"qe_stagnant_severe_consecutive=" f"{self.state.consecutive_qe_stagnant_severe}" ) logger.error( f"[som_auto_adjust] STOP: {self.state.stop_reason}" ) else: # Mild — boost leve self.state.alpha0_boost = min(2.0, self.state.alpha0_boost + 0.05) actions.append("qe_stagnant_mild_boost: α+0.05") else: self.state.consecutive_qe_stagnant_severe = 0 # ============================================================== # 2.4 Neighborhood Crossing → -α₀ (estabiliza), +σ₀ # V7 — Crossing severo consecutivo → stop-learning (Prova 12) # ============================================================== if crossing_detected: # Desacelera aprendizado para estabilizar cross_factor = 0.2 if crossing_severity == "severe" else 0.1 self.state.alpha0_boost = max(1.0, self.state.alpha0_boost - cross_factor) # Expande vizinhança para corrigir topologia self.state.sigma0_boost = min(2.0, self.state.sigma0_boost + 0.15) actions.append( f"neighborhood_crossing_adjust: α-{cross_factor:.2f}, σ+0.15 " f"({crossing_severity})" ) # V7 — Contador de crossing severo consecutivo if crossing_severity == "severe": self.state.consecutive_crossing_severe += 1 actions.append( f"crossing_severe_consecutive=" f"{self.state.consecutive_crossing_severe} " f"(limit={self.CROSSING_SEVERE_LIMIT})" ) # Intervenção extra em crossing severo: boost agressivo de σ # para forçar re-organização topológica. self.state.sigma0_boost = min(2.0, self.state.sigma0_boost + 0.25) actions.append( "crossing_severe_extra_boost: σ+0.25 (re-organização forçada)" ) if self.state.consecutive_crossing_severe >= self.CROSSING_SEVERE_LIMIT: self.state.stop_training = True self.state.stop_reason = ( f"neighborhood_crossing_severe_consecutive=" f"{self.state.consecutive_crossing_severe}" ) logger.error( f"[som_auto_adjust] STOP: {self.state.stop_reason} " f"(instabilidade sináptica crônica por " f"{self.state.consecutive_crossing_severe} batches)" ) else: # Crossing mild: reseta contador de severo (mild é recuperável) self.state.consecutive_crossing_severe = 0 else: # Sem crossing: reseta contador self.state.consecutive_crossing_severe = 0 # ============================================================== # Aplica boosts efetivos ao SOM (se alterados) # ============================================================== if (self.state.alpha0_boost, self.state.sigma0_boost) != boosts_before: self._apply_boosts(kls) actions.append( f"boosts_applied: α₀_eff={kls.som.alpha0 * self.state.alpha0_boost:.4f}, " f"σ₀_eff={kls.som.sigma0 * self.state.sigma0_boost:.4f}" ) # ============================================================== # Decay suave dos boosts quando tudo está saudável (sem ações) # ============================================================== if not actions and not self.state.stop_training: self.state.alpha0_boost = max(1.0, self.state.alpha0_boost * 0.98) self.state.sigma0_boost = max(1.0, self.state.sigma0_boost * 0.98) # Reset counters em batch saudável self.state.reset_consecutive_counters() # ============================================================== # Persiste ação no log # ============================================================== action_record = { "batch_idx": int(batch_idx), "actions": actions, "metrics_snapshot": { "qe": qe, "te": te, "kl": kl, "ve": ve, "dead_rate": dead_rate, "collapse_severity": collapse_severity, "qe_stagnant": qe_stagnant, "crossing_detected": crossing_detected, "nan_detected": nan_detected, }, "boosts": { "alpha0_boost": float(self.state.alpha0_boost), "sigma0_boost": float(self.state.sigma0_boost), }, "stop_training": bool(self.state.stop_training), } self.state.actions_log.append(action_record) if len(self.state.actions_log) > 50: self.state.actions_log = self.state.actions_log[-50:] if actions: self.state.n_actions_total += len(actions) return { "actions": actions, "stop_training": bool(self.state.stop_training), "stop_reason": self.state.stop_reason, "state": self.state.to_dict(), "boosts": { "alpha0_boost": float(self.state.alpha0_boost), "sigma0_boost": float(self.state.sigma0_boost), }, "n_actions": int(len(actions)), } # ------------------------------------------------------------------ # Helpers internos # ------------------------------------------------------------------ def _apply_boosts(self, kls: Any) -> None: """Aplica os boosts efetivos ao SOM. Math: α₀_eff = α₀_canonical · alpha0_boost σ₀_eff = σ₀_canonical · sigma0_boost O boost é multiplicativo sobre o α₀/σ₀ canônicos (FIX-3: sempre ≥ 1.0, nunca abaixo do canônico para não degradar abaixo do baseline). O decaimento temporal (α_t = α₀_eff · exp(-t/τ)) ainda aplica normalmente. """ try: som = kls.som # Boost afeta o α₀ e σ₀ que são usados em update_weights # α₀_eff é multiplicado internamente. Como KohonenSOM4D usa # self.alpha0 diretamente em update_weights, ajustamos um # campo _eff separado para preservar o canônico. if not hasattr(som, '_alpha0_canonical'): som._alpha0_canonical = float(som.alpha0) som._sigma0_canonical = float(som.sigma0) som.alpha0 = float(som._alpha0_canonical * self.state.alpha0_boost) som.sigma0 = float(som._sigma0_canonical * self.state.sigma0_boost) # Clamp para segurança (never below canonical, never above 2×) som.alpha0 = max(som._alpha0_canonical, min(2.0 * som._alpha0_canonical, som.alpha0)) som.sigma0 = max(som._sigma0_canonical, min(2.0 * som._sigma0_canonical, som.sigma0)) except Exception as e: logger.warning(f"[som_auto_adjust] _apply_boosts failed: {e}") def _no_action_response(self, batch_idx: int, reason: str) -> Dict[str, Any]: return { "actions": [], "stop_training": bool(self.state.stop_training), "stop_reason": self.state.stop_reason, "state": self.state.to_dict(), "boosts": { "alpha0_boost": float(self.state.alpha0_boost), "sigma0_boost": float(self.state.sigma0_boost), }, "n_actions": 0, "skip_reason": reason, } # ============================================================================ # 3. Helper para integração rápida no train script # ============================================================================ def create_auto_adjuster() -> SOMAutoAdjuster: """Factory para criar um SOMAutoAdjuster com estado novo.""" return SOMAutoAdjuster(state=SOMAutoAdjustState()) __all__ = [ "SOMAutoAdjustState", "SOMAutoAdjuster", "create_auto_adjuster", ]