| """Adaptive entropy + ruggedness phase controller for AHD-CMA. |
| |
| Three modes are emitted by :meth:`AdaptivePhaseController.select_mode`: |
| |
| * ``"explore"`` — high entropy *or* high ruggedness; the swarm is well |
| spread but the landscape is rough, so the DOA branch dominates. |
| * ``"exploit"`` — low entropy *or* very smooth landscape; collapse the |
| search via CMA-ES. |
| * ``"hybrid"`` — neither extreme dominates; mix both branches. |
| |
| Thresholds ``tau_low`` and ``tau_high`` are *adaptive*: every |
| ``threshold_update_interval`` generations the controller re-fits them |
| from the running history of entropy values, using |
| ``tau_high = mean + tau_high_offset * std`` and likewise for low. |
| |
| The mode rule is:: |
| |
| if entropy >= tau_high: explore |
| elif entropy <= tau_low: exploit |
| elif ruggedness >= ruggedness_exploit: explore # rugged -> need DOA |
| elif ruggedness <= ruggedness_explore: exploit # smooth -> need CMA-ES |
| else: hybrid |
| |
| Rationale: standard fitness-landscape analysis (Weinberger 1990; Pitzer |
| & Affenzeller 2012) maps high lag-1 autocorrelation -> smooth, low -> |
| rugged. Smooth landscapes benefit from CMA-ES's covariance learning, |
| rugged landscapes benefit from DOA's global perturbation. Empirically |
| on Rastrigin 30D the ruggedness signal saturates at ~1.0, so the |
| threshold names ``ruggedness_explore`` (lower bound) and |
| ``ruggedness_exploit`` (upper bound) here name the *modes their branch |
| emits*: rugg below ``ruggedness_explore`` -> exploit; rugg above |
| ``ruggedness_exploit`` -> explore. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from typing import Any, Literal |
|
|
| import numpy as np |
|
|
| Mode = Literal["explore", "exploit", "hybrid"] |
|
|
|
|
| class AdaptivePhaseController: |
| """Stateful controller that maps (entropy, ruggedness, t) → mode.""" |
|
|
| def __init__(self, config: dict[str, Any]) -> None: |
| self.entropy_bins: int = int(config.get("entropy_bins", 10)) |
| self.walk_length: int = int(config.get("ruggedness_walk_length", 10)) |
| self.update_interval: int = int(config.get("threshold_update_interval", 5)) |
| self.tau_high_offset: float = float(config.get("tau_high_offset", 0.5)) |
| self.tau_low_offset: float = float(config.get("tau_low_offset", -0.5)) |
| self.rugged_explore: float = float(config.get("ruggedness_explore_thresh", 0.3)) |
| self.rugged_exploit: float = float(config.get("ruggedness_exploit_thresh", 0.6)) |
|
|
| if self.tau_low_offset >= self.tau_high_offset: |
| raise ValueError( |
| f"tau_low_offset ({self.tau_low_offset}) must be strictly below " |
| f"tau_high_offset ({self.tau_high_offset})" |
| ) |
| if self.rugged_explore >= self.rugged_exploit: |
| raise ValueError( |
| f"ruggedness_explore_thresh ({self.rugged_explore}) must be " |
| f"strictly below ruggedness_exploit_thresh ({self.rugged_exploit})" |
| ) |
|
|
| self._entropy_history: list[float] = [] |
| self.tau_low: float = -np.inf |
| self.tau_high: float = np.inf |
|
|
| def update_thresholds(self, history: list[float]) -> None: |
| """Re-fit ``tau_low`` and ``tau_high`` from a list of entropy values.""" |
| if len(history) < 2: |
| return |
| arr = np.asarray(history, dtype=np.float64) |
| mu = float(arr.mean()) |
| sigma = float(arr.std()) |
| self.tau_high = mu + self.tau_high_offset * sigma |
| self.tau_low = mu + self.tau_low_offset * sigma |
|
|
| def select_mode(self, entropy: float, ruggedness: float, t: int) -> Mode: |
| """Decide the mode for generation ``t`` and update internal state.""" |
| self._entropy_history.append(float(entropy)) |
| if t > 0 and t % self.update_interval == 0: |
| self.update_thresholds(self._entropy_history) |
|
|
| if np.isfinite(self.tau_high) and entropy >= self.tau_high: |
| return "explore" |
| if np.isfinite(self.tau_low) and entropy <= self.tau_low: |
| return "exploit" |
| if ruggedness >= self.rugged_exploit: |
| |
| return "explore" |
| if ruggedness <= self.rugged_explore: |
| |
| return "exploit" |
| return "hybrid" |
|
|
| @property |
| def entropy_history(self) -> list[float]: |
| """Read-only view of the recorded entropy values (in order).""" |
| return list(self._entropy_history) |
|
|
