GeneticWFM / src /utils /health.py
GaetanoParente's picture
first commit
9e62f55
from numba import njit
# --- 1. METRICHE PRE-FLIGHT: Indice di Bilanciamento Evolutivo (IBE) ---
def calculate_ibe(pop_size, generations, p_cross, p_mut, p_heur, p_elite):
"""
Calcola l'Indice di Bilanciamento Evolutivo (IBE), una metrica euristica
per stimare il trade-off tra esplorazione (crossover/mutazione) e
sfruttamento o pressione selettiva (euristiche/elitismo).
Target empirico di stabilità: 1000 - 3000.
"""
# Energia cinetica (Generazione di nuova varianza)
numerator = (p_cross * pop_size) + (p_mut * pop_size * generations)
# Fattori frenanti (Scaling 0-100 per bilanciare l'ordine di grandezza del numeratore)
h_val_score = max(p_heur * 100.0, 1.0)
e_val_score = max(p_elite * 100.0, 0.1)
denominator = h_val_score * e_val_score
return numerator / denominator
def interpret_ibe(ibe_value):
"""Valuta il setup dei parametri rispetto al regime di stabilità ottimale."""
if 1000 <= ibe_value <= 3000:
return "OTTIMALE (Bilanciato)", "normal"
elif ibe_value < 1000:
return "RISCHIO STALLO (Eccessiva pressione selettiva)", "off"
else:
return "RISCHIO CAOS (Esplorazione incontrollata)", "inverse"
# --- 2. METRICHE POST-FLIGHT: Distanza di Hamming & Convergenza ---
@njit(cache=True)
def calculate_diversity_snapshot(population):
"""
Calcola la diversità genetica della popolazione tramite Distanza di Hamming.
Implementazione JIT con campionamento stocastico per evitare l'overhead O(N^2).
"""
pop_size, num_emps, num_days = population.shape
if pop_size < 2: return 0.0
# Clamp del sample size per limitare il costo computazionale
sample_size = min(50, pop_size)
total_diffs = 0
total_comparisons = 0
genome_len = num_emps * num_days
# Valutazione differenziale cross-individuo
for i in range(sample_size):
for j in range(i + 1, pop_size):
diff_count = 0
for e in range(num_emps):
for d in range(num_days):
if population[i, e, d] != population[j, e, d]:
diff_count += 1
total_diffs += diff_count
total_comparisons += 1
if total_comparisons == 0: return 0.0
avg_diff = total_diffs / total_comparisons
# Normalizzazione % rispetto allo spazio di ricerca del singolo fenotipo
diversity_percentage = (avg_diff / genome_len) * 100.0
return diversity_percentage
def analyze_convergence_quality(history):
"""
Analizza la time-series della diversità per classificare il regime di convergenza
(Matura, Prematura o Divergente).
"""
if not history:
return "Dati insufficienti", "off"
final_diversity = history[-1]
# 1. Divergenza di sistema (Assenza di convergenza locale o globale)
if final_diversity > 40.0:
return "⚠️ CRITICO: Caos (Assenza di convergenza)", "inverse"
# 2. Regime esplorativo ancora attivo
if final_diversity >= 5.0:
return f"✅ SANO: Diversità Attiva ({final_diversity:.2f}%)", "normal"
# 3. Analisi del drop-rate per rilevare 'Premature Convergence' (Collasso genotipico)
threshold_idx = -1
for i, val in enumerate(history):
if val < 5.0:
threshold_idx = i
break
total_steps = len(history)
# Se la diversità fisiologica si è mantenuta intatta fino all'ultimo delta
if threshold_idx == -1:
return "✅ SANO: Convergenza in corso", "normal"
# Classificazione basata sull'epoca del collasso (< 20% delle generazioni totali = prematuro)
if threshold_idx < (total_steps * 0.20):
return "⚠️ CRITICO: Convergenza Prematura (Collasso genotipico)", "inverse"
else:
return "🏆 ECCELLENTE: Convergenza Matura (Consenso Raggiunto)", "success"