From 1671650b2b6d723999a0b84a822a493bb2a7c400 Mon Sep 17 00:00:00 2001 From: Ari Aye Date: Sun, 30 Aug 2026 13:45:50 -0700 Subject: [PATCH] feat: auto-switch to rank-weighted selection based on archive size sample_parent() now automatically enables rank-weighted tournament selection when the archive reaches auto_rank_cell_threshold (default 8) occupied cells, indicating enough diversity that biasing toward stronger parents is beneficial. This mirrors the on_plateau() pattern of adapting strategy when the search state calls for it. On by default; opt out with auto_rank_weighted=False. Co-Authored-By: Claude Opus 4.6 (1M context) --- factory/outer_loop/population.py | 15 ++++++++- tests/test_outer_loop/test_population.py | 42 +++++++++++++++++++++++- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/factory/outer_loop/population.py b/factory/outer_loop/population.py index 451ec6e0..4022c9d9 100644 --- a/factory/outer_loop/population.py +++ b/factory/outer_loop/population.py @@ -129,6 +129,8 @@ def sample_parent( self, tournament_size: int = 3, rank_weighted: bool = False, + auto_rank_weighted: bool = True, + auto_rank_cell_threshold: int = 8, ) -> Individual | None: """Tournament selection: pick tournament_size individuals, return the best. @@ -136,6 +138,12 @@ def sample_parent( proportional to their rank (best=N, worst=1) instead of uniformly. This biases toward stronger parents while still allowing weaker individuals a small chance, preserving diversity. + + When ``auto_rank_weighted=True`` (default), rank-weighted selection + activates automatically when the archive reaches + ``auto_rank_cell_threshold`` occupied cells, mirroring + ``on_plateau()``'s pattern of adapting strategy when the search + state calls for it. """ import random @@ -143,7 +151,12 @@ def sample_parent( if not individuals: return None k = min(tournament_size, len(individuals)) - if rank_weighted and len(individuals) >= 2: + + use_rank = rank_weighted + if not use_rank and auto_rank_weighted and len(individuals) >= auto_rank_cell_threshold: + use_rank = True + + if use_rank and len(individuals) >= 2: ranked = sorted(individuals, key=lambda i: i.score) weights = [rank + 1.0 for rank in range(len(ranked))] tournament = random.choices(ranked, weights=weights, k=k) diff --git a/tests/test_outer_loop/test_population.py b/tests/test_outer_loop/test_population.py index 8991183c..6c024ffb 100644 --- a/tests/test_outer_loop/test_population.py +++ b/tests/test_outer_loop/test_population.py @@ -198,9 +198,49 @@ def test_rank_weighted_false_is_uniform(self) -> None: archive.add(Individual(id="b", workflow_data={}, score=1000.0, features=(1, 0, 1, 0))) counts: dict[str, int] = {"a": 0, "b": 0} for _ in range(200): - p = archive.sample_parent(tournament_size=1, rank_weighted=False) + p = archive.sample_parent(tournament_size=1, rank_weighted=False, auto_rank_weighted=False) assert p is not None counts[p.id] += 1 # Uniform: both should be roughly 50/50 assert counts["a"] > 50 assert counts["b"] > 50 + + +class TestAutoRankWeighted: + def test_auto_activates_by_cell_count(self) -> None: + archive = MAPElitesArchive() + for i in range(10): + archive.add(Individual(id=f"i{i}", workflow_data={}, score=i * 10.0, features=(i,))) + counts: dict[str, int] = {} + for _ in range(300): + p = archive.sample_parent(tournament_size=1, auto_rank_cell_threshold=8) + assert p is not None + counts[p.id] = counts.get(p.id, 0) + 1 + # With auto rank-weighted, best should be picked more often + assert counts.get("i9", 0) > counts.get("i0", 0) + + def test_auto_stays_uniform_below_thresholds(self) -> None: + archive = MAPElitesArchive() + archive.add(Individual(id="a", workflow_data={}, score=10.0, features=(0,))) + archive.add(Individual(id="b", workflow_data={}, score=11.0, features=(1,))) + # 2 cells < 8 threshold, variance ~0.25 < 1000 threshold + counts: dict[str, int] = {"a": 0, "b": 0} + for _ in range(200): + p = archive.sample_parent(tournament_size=1) + assert p is not None + counts[p.id] += 1 + assert counts["a"] > 50 + assert counts["b"] > 50 + + def test_auto_disabled(self) -> None: + archive = MAPElitesArchive() + for i in range(10): + archive.add(Individual(id=f"i{i}", workflow_data={}, score=i * 100.0, features=(i,))) + counts: dict[str, int] = {} + for _ in range(300): + p = archive.sample_parent(tournament_size=1, auto_rank_weighted=False) + assert p is not None + counts[p.id] = counts.get(p.id, 0) + 1 + # Without auto, uniform sampling — worst should get ~10% (1/10) + assert counts.get("i0", 0) > 10 +