core/meta_meta_optimizer.py

# core/meta_meta_optimizer.py
import math
import shutil
import logging
import importlib.util
from pathlib import Path
from typing import Dict, Any

log = logging.getLogger("vortex.meta_meta")

# Fonctions de benchmark (pour évaluer FellowOptimizer)
BENCH_FUNCS = [
    ("sphere", lambda p: sum(x**2 for x in p), [(-5.12, 5.12)] * 5, 0.0),
    ("rastrigin", lambda p: 10 * len(p) + sum(xi**2 - 10 * math.cos(2 * math.pi * xi) for xi in p),
     [(-5.12, 5.12)] * 5, 0.0),
    ("rosenbrock", lambda p: sum(100 * (p[i + 1] - p[i]**2)**2 + (1 - p[i])**2 for i in range(len(p) - 1)),
     [(-2.0, 2.0)] * 5, 0.0),
]


class MetaMetaOptimizer:
    def __init__(self, kernel, llm_engine, current_optimizer_path: str):
        self.kernel = kernel
        self.llm = llm_engine
        self.path = Path(current_optimizer_path)
        self.backup = self.path.with_suffix(".py.bak")
        self.proof_system = None  # sera injecté plus tard
        self.status = {
            "last_improvement": None,
            "regression_count": 0,
            "best_score": float('inf'),
            "current_score": float('inf'),
            "benchmark_results": {}
        }

    # ─── Injection du système de preuves ────────────────────────────────
    def set_proof_system(self, proof_system):
        """Lie le ProofSystem pour enregistrer les améliorations."""
        self.proof_system = proof_system
        # Charger le meilleur score historique
        if self.proof_system and self.proof_system.proofs:
            best = min(p.get("after", float('inf')) for p in self.proof_system.proofs)
            self.status["best_score"] = best

    # ─── Benchmark ────────────────────────────────────────────────────────
    def _bench_module(self, mod) -> float:
        """Évalue un module FellowOptimizer sur les 3 fonctions de benchmark."""
        try:
            Optimizer = getattr(mod, "FellowOptimizer")
        except AttributeError:
            return float('inf')

        total_score = 0.0
        results = {}
        for name, func, bounds, opt_v in BENCH_FUNCS:
            try:
                optimizer = Optimizer(func, bounds, max_evals=100)
                best, _ = optimizer.optimize()
                error = abs(func(best) - opt_v)
                error = min(error, 1000.0)
                score = 1.0 - math.tanh(error / 50.0)
                total_score += score
                results[name] = round(score, 3)
            except Exception as e:
                log.error(f"[META-META] Erreur benchmark {name}: {e}")
                results[name] = 0.0
                total_score += 0.0

        avg_score = total_score / len(BENCH_FUNCS) if BENCH_FUNCS else 0.0
        self.status["benchmark_results"] = results
        return round(avg_score, 3)

    def _bench_file(self, filepath: Path) -> float:
        """Charge un fichier Python et le benchmark."""
        try:
            spec = importlib.util.spec_from_file_location("_optimizer_bench", filepath)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            return self._bench_module(mod)
        except Exception as e:
            log.error(f"[META-META] Erreur chargement {filepath}: {e}")
            return float('inf')

    # ─── Cycle d'amélioration ────────────────────────────────────────────
    async def monitor_and_improve(self) -> Dict[str, Any]:
        """Mesure la performance de l'optimiseur actuel et l'améliore si possible."""
        log.info("[META-META] Démarrage du benchmark...")

        # 1. Benchmark du code actuel
        current_score = self._bench_file(self.path)
        self.status["current_score"] = current_score
        log.info(f"[META-META] Score actuel : {current_score:.4f}")

        # 2. Comparer avec le meilleur score historique
        best_score = self.status.get("best_score", float('inf'))
        result = {
            "action": "none",
            "score": current_score,
            "improvement": 0.0
        }

        if self.proof_system is not None:
            version = f"v{len(self.proof_system.proofs) + 1}"
            before = best_score if best_score != float('inf') else current_score
            after = current_score

            if after < before - 0.005:  # Amélioration > 0.5%
                self.proof_system.register_improvement(
                    version=version,
                    before=before,
                    after=after,
                    benchmark="FellowOptimizer (Sphere/Rastrigin/Rosenbrock)"
                )
                self.status["best_score"] = after
                self.status["last_improvement"] = version
                result["action"] = "improved"
                result["improvement"] = before - after
                log.info(f"[META-META] ✅ Amélioration ! {before:.4f} → {after:.4f}")
            else:
                # Enregistrement pour suivi
                self.proof_system.register_improvement(
                    version=version,
                    before=before,
                    after=after,
                    benchmark="FellowOptimizer (monitoring)"
                )
                if after > before:
                    self.status["regression_count"] += 1
                    result["action"] = "regression"
                    result["improvement"] = after - before
                    log.warning(f"[META-META] ⚠️ Régression : {before:.4f} → {after:.4f}")
                else:
                    result["action"] = "stable"
                    log.info(f"[META-META] Stable : {before:.4f} → {after:.4f}")

            self.status["current_score"] = after

        return result

    def rollback(self) -> bool:
        """Restaure le fichier optimiseur depuis la sauvegarde."""
        if not self.backup.exists():
            log.warning("[META-META] Aucun backup disponible.")
            return False
        try:
            shutil.copy(self.backup, self.path)
            log.info("[META-META] Rollback effectué avec succès.")
            return True
        except Exception as e:
            log.error(f"[META-META] Rollback échoué : {e}")
            return False

    def get_status(self) -> Dict:
        return self.status