core/hypothesis_tree.py

# core/hypothesis_tree.py
"""
Arbre d'Hypothèses – exploration parallèle de stratégies.
Chaque nœud est une hypothèse (description, code, score, statut).
L'arbre gère l'élagage et la capitalisation des échecs via Harbour.
"""
import logging
import time
import json
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from enum import Enum

logger = logging.getLogger("vortex.hypothesis_tree")

class BranchStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    PRUNED = "pruned"

@dataclass
class Branch:
    id: str
    hypothesis: str          # description de la stratégie
    code: str                # code à exécuter (si applicable)
    score: float = 0.0
    status: BranchStatus = BranchStatus.PENDING
    result: Optional[Any] = None
    error: Optional[str] = None
    execution_time: float = 0.0
    children: List['Branch'] = field(default_factory=list)
    parent_id: Optional[str] = None

class HypothesisTree:
    """
    Gère un ensemble de branches (hypothèses) pour une tâche donnée.
    Permet l'ajout, l'exécution en parallèle, l'élagage et le reporting.
    """

    def __init__(self, task: str, max_branches: int = 5, max_depth: int = 2):
        self.task = task
        self.max_branches = max_branches
        self.max_depth = max_depth
        self.root: Optional[Branch] = None
        self.branches: Dict[str, Branch] = {}
        self._next_id = 0

    def add_branch(self, hypothesis: str, code: str = "", parent_id: Optional[str] = None) -> str:
        """Ajoute une nouvelle branche à l'arbre."""
        branch_id = f"branch_{self._next_id}"
        self._next_id += 1
        branch = Branch(
            id=branch_id,
            hypothesis=hypothesis,
            code=code,
            status=BranchStatus.PENDING,
            parent_id=parent_id
        )
        self.branches[branch_id] = branch

        if parent_id is None:
            self.root = branch
        else:
            parent = self.branches.get(parent_id)
            if parent:
                parent.children.append(branch)

        logger.info(f"[HypothesisTree] Branche ajoutée: {branch_id} - {hypothesis[:60]}...")
        return branch_id

    def update_branch(self, branch_id: str, status: BranchStatus, score: float = 0.0,
                      result: Any = None, error: str = None, execution_time: float = 0.0):
        """Met à jour le statut et les métriques d'une branche."""
        branch = self.branches.get(branch_id)
        if not branch:
            return
        branch.status = status
        branch.score = score
        branch.result = result
        branch.error = error
        branch.execution_time = execution_time
        logger.debug(f"[HypothesisTree] Mise à jour {branch_id}: {status.value} (score={score:.2f})")

    def get_best_branch(self) -> Optional[Branch]:
        """Retourne la branche avec le score le plus élevé (parmi SUCCESS)."""
        best = None
        for b in self.branches.values():
            if b.status == BranchStatus.SUCCESS and (best is None or b.score > best.score):
                best = b
        return best

    def get_summary(self) -> Dict:
        """Retourne un résumé de l'arbre."""
        return {
            "task": self.task,
            "total_branches": len(self.branches),
            "successful": sum(1 for b in self.branches.values() if b.status == BranchStatus.SUCCESS),
            "failed": sum(1 for b in self.branches.values() if b.status == BranchStatus.FAILED),
            "pruned": sum(1 for b in self.branches.values() if b.status == BranchStatus.PRUNED),
            "pending": sum(1 for b in self.branches.values() if b.status == BranchStatus.PENDING),
            "best_score": self.get_best_branch().score if self.get_best_branch() else 0.0,
        }

    def prune_low_scoring(self, threshold: float = 0.5):
        """Élagage des branches avec score < seuil (sauf si c'est la seule)."""
        to_prune = []
        for b in self.branches.values():
            if b.status == BranchStatus.SUCCESS and b.score < threshold:
                to_prune.append(b.id)
        for bid in to_prune:
            self.branches[bid].status = BranchStatus.PRUNED
            logger.info(f"[HypothesisTree] Branche élaguée: {bid} (score trop bas)")