From c676a86d0d98dd8ba7b3c6b8800b5a9ccc2c0bda Mon Sep 17 00:00:00 2001 From: Nick Date: Fri, 13 Nov 2026 11:23:07 -0600 Subject: [PATCH] Add resumable Bayesian TPE optimization for TuneX Implement a dependency-free discrete Tree-structured Parzen Estimator that proposes integer UCI tuning vectors using good/bad outcome densities and a smoothed l/g acquisition objective. Evaluate paired matches against a fixed baseline, track trial WDL and objectives, persist reproducible resume state, reject mismatched configurations, and expose a bayesian CLI with sample config and regression tests. --- README.md | 40 ++++++ configs/bayesian.example.json | 72 ++++++++++ src/tunex/bayesian.py | 245 ++++++++++++++++++++++++++++++++++ src/tunex/cli.py | 15 +++ src/tunex/config.py | 57 ++++++++ tests/test_bayesian.py | 148 ++++++++++++++++++++ 6 files changed, 577 insertions(+) create mode 100644 configs/bayesian.example.json create mode 100644 src/tunex/bayesian.py create mode 100644 tests/test_bayesian.py diff --git a/README.md b/README.md index b80f2cd..4b4c7cd 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,46 @@ test results reflect SpellWarden's current rules implementation. IronPhoenix four-player behavior, FEN4/SAN4 parsing, team pairing, and all existing configs remain unchanged when `variant` is omitted. +## Bayesian optimization (TPE) + +TuneX supports a dependency-free **Tree-structured Parzen Estimator (TPE)** +for model-guided optimization of discrete/integer UCI parameters. It samples +initial configurations at random and uses smoothed density estimates from the +best and worst observed trials to choose promising combinations. TPE is a +Bayesian optimization approach, not a Gaussian-process surrogate. + +Copy `configs/bayesian.example.json`, adjust `engine.path` and the options +you want to tune, then run: + +~~~powershell +tunex -c configs/bayesian.example.json bayesian +tunex -c configs/bayesian.example.json bayesian --iterations 150 +~~~ + +The `bayesian` section contains `iterations`, `games_per_trial` (even), +`startup_trials`, `candidates` (proposal pool size), `gamma` (fraction +of strongest observations), `prior_weight` (noise shrinkage toward 50%), +`seed`, `state_file`, and integer `parameters` with `name`, `value`, +`min`, `max`, `step`. The existing `baseline.options` are the fixed +control settings. TuneX tests each proposed candidate against the **same +baseline executable and options**, using paired team/color swaps and the +existing match runner. The first trial uses the supplied starting values. + +A completed trial is recorded with its exact parameter vector, W/D/L, and +score. Random state, all observations, and the best observed options are +saved atomically so a subsequent command resumes deterministically. A resume +with incompatible time controls, baseline, parameter ranges, seed or engine +path is rejected to avoid mixing unrelated experiments. Do not run multiple +tuners against one state file at the same time. + +This is an optimizer, **not a guarantee of Elo gains**. With small per-trial +game budgets, even a model-guided tuner can select lucky configurations. +Independently verify the reported best settings against an untouched baseline +using SPRT and fresh openings. The current optimizer supports integer-valued +spin options; boolean switches are best evaluated separately. It works with +IronPhoenix and SpellWarden by selecting the corresponding `variant` and +using parameters actually advertised by that engine. + ## Requirements - Python **3.11+** diff --git a/configs/bayesian.example.json b/configs/bayesian.example.json new file mode 100644 index 0000000..54ac76e --- /dev/null +++ b/configs/bayesian.example.json @@ -0,0 +1,72 @@ +{ + "variant": "four_player_teams", + "engine": { + "path": "C:/path/to/IronPhoenix/out/build/x64-release/ironphoenix.exe", + "startup_timeout": 10, + "move_timeout": 60, + "strict_options": true, + "common_options": { + "Hash": 64, + "MultiPV": 1, + "UseNNUE": false + } + }, + "search": { + "mode": "movetime", + "value": 250 + }, + "match": { + "games": 100, + "workers": 4, + "max_plies": 320, + "paired": true, + "seed": 20261113, + "openings_file": null, + "results_file": "results/bayesian_games.jsonl" + }, + "baseline": { + "name": "Fixed IronPhoenix baseline", + "options": { + "SearchLMRBaseX100": 79, + "SearchLMRDivisorX100": 300, + "SearchSEEPruneMargin": 87 + } + }, + "candidate": { + "name": "Bayesian candidate", + "options": {} + }, + "bayesian": { + "iterations": 100, + "games_per_trial": 40, + "startup_trials": 12, + "candidates": 64, + "gamma": 0.25, + "prior_weight": 4, + "seed": 20261113, + "state_file": "results/ironphoenix_bayesian_state.json", + "parameters": [ + { + "name": "SearchLMRBaseX100", + "value": 79, + "min": 50, + "max": 120, + "step": 2 + }, + { + "name": "SearchLMRDivisorX100", + "value": 300, + "min": 220, + "max": 420, + "step": 5 + }, + { + "name": "SearchSEEPruneMargin", + "value": 87, + "min": 40, + "max": 160, + "step": 5 + } + ] + } +} diff --git a/src/tunex/bayesian.py b/src/tunex/bayesian.py new file mode 100644 index 0000000..b141d8d --- /dev/null +++ b/src/tunex/bayesian.py @@ -0,0 +1,245 @@ +"""Dependency-free Bayesian optimization using a discrete TPE surrogate. + +A Tree-structured Parzen Estimator (TPE) fits independent smoothed +one-dimensional densities to good and bad observed configurations, then +samples the good distribution and ranks proposals by log(l(x) / g(x)). +This is a sequential, model-guided optimizer, not Gaussian-process BO. +""" +from __future__ import annotations + +from dataclasses import asdict +import json +import math +import os +from pathlib import Path +import random +from typing import Any + +from .config import BayesianConfig, Config, SpsaParameter, VariantConfig +from .match import MatchRunner +from .stats import format_summary + + +def _grid(param: SpsaParameter) -> list[int]: + low, high, step = int(param.minimum), int(param.maximum), int(param.step) + return list(range(low, high + 1, step)) + + +def _nearest(value: float, param: SpsaParameter) -> int: + values = _grid(param) + return min(values, key=lambda v: (abs(value - v), v)) + + +def _signature(config: Config, tune: BayesianConfig) -> str: + # Reject incompatible resume attempts rather than silently mixing matches + # from different engines, time controls, baselines, or parameter spaces. + payload: dict[str, Any] = { + "variant": config.variant, + "engine": asdict(config.engine), + "search": asdict(config.search), + "match": { + "max_plies": config.match.max_plies, + "paired": config.match.paired, + "seed": config.match.seed, + "openings_file": config.match.openings_file, + }, + "baseline": asdict(config.baseline), + "games_per_trial": tune.games_per_trial, + "startup_trials": tune.startup_trials, + "candidates": tune.candidates, + "gamma": tune.gamma, + "prior_weight": tune.prior_weight, + "parameters": [asdict(p) for p in tune.parameters], + "seed": tune.seed, + } + return json.dumps(payload, sort_keys=True) + + +def _density(value: int, observations: list[int], param: SpsaParameter) -> float: + """Smoothed mixture of Gaussian kernels and a uniform prior on grid.""" + grid = _grid(param) + spread = max(float(param.step), (param.maximum - param.minimum) * 0.12) + if not observations: + return 1.0 / len(grid) + # Truncate/normalize on the finite grid to avoid boundary bias. + probabilities = [] + for observed in observations: + weights = [math.exp(-0.5 * ((g - observed) / spread) ** 2) for g in grid] + total = sum(weights) + probabilities.append(weights[grid.index(value)] / total) + return 0.1 / len(grid) + 0.9 * sum(probabilities) / len(probabilities) + + +def _draw_good( + rng: random.Random, observations: list[int], param: SpsaParameter +) -> int: + grid = _grid(param) + if not observations: + return rng.choice(grid) + observed = rng.choice(observations) + spread = max(float(param.step), (param.maximum - param.minimum) * 0.12) + if rng.random() < 0.10: + return rng.choice(grid) + return _nearest(rng.gauss(observed, spread), param) + + +def _choose( + rng: random.Random, + history: list[dict[str, Any]], + config: BayesianConfig, +) -> dict[str, int]: + params = config.parameters + previous = { + tuple(int(trial["options"][p.name]) for p in params) + for trial in history + } + capacity = math.prod(len(_grid(p)) for p in params) + if len(previous) >= capacity: + raise ValueError("Bayesian parameter search space is exhausted") + + def new(options: dict[str, int]) -> bool: + return tuple(options[p.name] for p in params) not in previous + + if not history: + first = {p.name: _nearest(p.value, p) for p in params} + if new(first): + return first + + if len(history) < config.startup_trials: + for _ in range(1000): + choice = {p.name: rng.choice(_grid(p)) for p in params} + if new(choice): + return choice + else: + ranked = sorted(history, key=lambda t: t["objective"], reverse=True) + elite = max(1, min(len(ranked) - 1, math.ceil(config.gamma * len(ranked)))) + good, bad = ranked[:elite], ranked[elite:] + best: dict[str, int] | None = None + best_score = -math.inf + for _ in range(max(config.candidates * 4, 128)): + proposal = { + p.name: _draw_good( + rng, [int(t["options"][p.name]) for t in good], p + ) + for p in params + } + if not new(proposal): + continue + ratio = 0.0 + for p in params: + x = proposal[p.name] + ratio += math.log(_density( + x, [int(t["options"][p.name]) for t in good], p + )) + ratio -= math.log(_density( + x, [int(t["options"][p.name]) for t in bad], p + )) + if ratio > best_score: + best, best_score = proposal, ratio + if best is not None: + return best + + # Guaranteed finite-grid fallback; also handles duplicate proposals. + for _ in range(1000): + candidate = {p.name: rng.choice(_grid(p)) for p in params} + if new(candidate): + return candidate + # Rare near-exhaustion case: enumerate only until an unused combination. + import itertools + for values in itertools.product(*(_grid(p) for p in params)): + if values not in previous: + return dict(zip((p.name for p in params), values)) + raise ValueError("Bayesian parameter search space is exhausted") + + +class BayesianRunner: + def __init__(self, config: Config) -> None: + if config.bayesian is None: + raise ValueError("config has no bayesian section") + self.config = config + self.tune = config.bayesian + if not config.match.paired: + raise ValueError("Bayesian tuning requires paired matches") + self.match_runner = MatchRunner(config) + self.rng = random.Random(self.tune.seed) + self.trials: list[dict[str, Any]] = [] + self._load_state() + + @property + def state_path(self) -> Path: + return Path(self.tune.state_file) + + def _load_state(self) -> None: + path = self.state_path + if not path.exists(): + return + state = json.loads(path.read_text(encoding="utf-8")) + if state.get("signature") != _signature(self.config, self.tune): + raise ValueError( + f"Bayesian state {path} belongs to different configuration; " + "use a new state_file for this experiment" + ) + self.trials = state["trials"] + self.rng.setstate(_tuple(state["rng_state"])) + + def _save_state(self) -> None: + path = self.state_path + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "version": 1, + "signature": _signature(self.config, self.tune), + "rng_state": self.rng.getstate(), + "trials": self.trials, + "best_options": self.best_options(), + } + temporary = path.with_name(path.name + ".tmp") + temporary.write_text(json.dumps(payload, indent=2), encoding="utf-8") + os.replace(temporary, path) + + def best_options(self) -> dict[str, int]: + if not self.trials: + return {p.name: _nearest(p.value, p) for p in self.tune.parameters} + winner = max(self.trials, key=lambda t: t["objective"]) + return dict(winner["options"]) + + def run(self, iterations: int | None = None) -> None: + target = self.tune.iterations if iterations is None else iterations + if target <= 0: + raise ValueError("Bayesian iterations must be positive") + for k in range(len(self.trials), target): + options = _choose(self.rng, self.trials, self.tune) + candidate_options = dict(self.config.baseline.options) + candidate_options.update(options) + # Match against a *fixed* baseline. The current best isn't + # a moving target; this preserves comparability of observations. + candidate = VariantConfig( + name="Bayesian candidate", options=candidate_options, + path=self.config.baseline.path, + ) + wdl, _ = self.match_runner.run( + baseline=self.config.baseline, + candidate=candidate, + games=self.tune.games_per_trial, + offset=k * self.tune.games_per_trial, + ) + # Shrink noisy small-sample match scores towards 50% to avoid + # over-trusting a lucky unbeaten trial. Kept for each observation. + score = ( + wdl.wins + 0.5 * wdl.draws + 0.5 * self.tune.prior_weight + ) / (wdl.games + self.tune.prior_weight) + trial = { + "iteration": k + 1, + "options": options, + "objective": score, + "wdl": asdict(wdl), + } + self.trials.append(trial) + self._save_state() + print(f"[Bayesian {k + 1}/{target}] {format_summary(wdl)}") + print(" evaluated:", options, "| best:", self.best_options()) + + +def _tuple(value: Any) -> Any: + if isinstance(value, list): + return tuple(_tuple(v) for v in value) + return value diff --git a/src/tunex/cli.py b/src/tunex/cli.py index 6a3c05e..4bb2668 100644 --- a/src/tunex/cli.py +++ b/src/tunex/cli.py @@ -5,6 +5,7 @@ import sys import threading +from .bayesian import BayesianRunner from .config import Config from .engine import EngineError, UciEngine from .match import GameRecord, GameSpec, MatchRunner @@ -168,6 +169,15 @@ def command_spsa(config: Config, iterations: int | None) -> int: return 0 +def command_bayesian(config: Config, iterations: int | None) -> int: + runner = BayesianRunner(config) + runner.run(iterations) + print("best observed Bayesian options (validate with SPRT):") + for name, value in runner.best_options().items(): + print(f" {name} = {value}") + return 0 + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="tunex", @@ -191,6 +201,9 @@ def build_parser() -> argparse.ArgumentParser: spsa = sub.add_parser("spsa", help="run SPSA parameter tuning") spsa.add_argument("--iterations", type=int, default=None) + bayesian = sub.add_parser("bayesian", help="run resumable Bayesian TPE tuning") + bayesian.add_argument("--iterations", type=int, default=None) + return parser @@ -208,6 +221,8 @@ def main(argv: list[str] | None = None) -> int: return command_sprt(config) if args.command == "spsa": return command_spsa(config, args.iterations) + if args.command == "bayesian": + return command_bayesian(config, args.iterations) except KeyboardInterrupt: print("\nTunex interrupted; active engine processes stopped.", file=sys.stderr) return 130 diff --git a/src/tunex/config.py b/src/tunex/config.py index 55f0faa..2bd05d6 100644 --- a/src/tunex/config.py +++ b/src/tunex/config.py @@ -197,6 +197,60 @@ def from_dict(cls, data: dict[str, Any] | None) -> "SpsaConfig": return cfg +@dataclass(slots=True) +class BayesianConfig: + """Sequential TPE-style Bayesian optimization of integer UCI options.""" + + iterations: int = 100 + games_per_trial: int = 40 + startup_trials: int = 12 + candidates: int = 64 + gamma: float = 0.25 + prior_weight: float = 4.0 + seed: int = 20261113 + state_file: str = "results/bayesian_state.json" + parameters: list[SpsaParameter] = field(default_factory=list) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "BayesianConfig": + cfg = cls( + iterations=int(data.get("iterations", 100)), + games_per_trial=int(data.get("games_per_trial", 40)), + startup_trials=int(data.get("startup_trials", 12)), + candidates=int(data.get("candidates", 64)), + gamma=float(data.get("gamma", 0.25)), + prior_weight=float(data.get("prior_weight", 4.0)), + seed=int(data.get("seed", 20261113)), + state_file=str(data.get("state_file", "results/bayesian_state.json")), + parameters=[SpsaParameter.from_dict(p) for p in data.get("parameters", [])], + ) + if cfg.iterations <= 0 or cfg.games_per_trial <= 0: + raise ValueError("bayesian iterations and games_per_trial must be positive") + if cfg.games_per_trial % 2: + raise ValueError("bayesian games_per_trial must be even for paired testing") + if cfg.startup_trials < 2 or cfg.candidates <= 0: + raise ValueError("bayesian startup_trials must be >= 2 and candidates > 0") + if not 0 < cfg.gamma <= 0.5: + raise ValueError("bayesian gamma must be in (0, 0.5]") + if cfg.prior_weight < 0: + raise ValueError("bayesian prior_weight must be >= 0") + if not cfg.state_file.strip(): + raise ValueError("bayesian state_file cannot be empty") + if not cfg.parameters: + raise ValueError("bayesian parameters must not be empty") + if len({p.name for p in cfg.parameters}) != len(cfg.parameters): + raise ValueError("bayesian parameter names must be unique") + for p in cfg.parameters: + if p.minimum == p.maximum: + raise ValueError(f"{p.name}: Bayesian bounds must allow variation") + if any(abs(x - round(x)) > 1e-9 for x in + (p.minimum, p.maximum, p.value, p.step)): + raise ValueError(f"{p.name}: Bayesian parameters must be integer-valued") + if p.step < 1: + raise ValueError(f"{p.name}: Bayesian step must be >= 1") + return cfg + + @dataclass(slots=True) class Config: engine: EngineConfig @@ -206,6 +260,7 @@ class Config: candidate: VariantConfig sprt: SprtConfig spsa: SpsaConfig | None = None + bayesian: BayesianConfig | None = None variant: str = "four_player_teams" @classmethod @@ -235,5 +290,7 @@ def load(cls, path: str | Path) -> "Config": candidate=candidate, sprt=SprtConfig.from_dict(data.get("sprt")), spsa=SpsaConfig.from_dict(data["spsa"]) if "spsa" in data else None, + bayesian=BayesianConfig.from_dict(data["bayesian"]) + if "bayesian" in data else None, variant=variant, ) diff --git a/tests/test_bayesian.py b/tests/test_bayesian.py new file mode 100644 index 0000000..74e0f99 --- /dev/null +++ b/tests/test_bayesian.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import json +import random +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from tunex.bayesian import BayesianRunner, _choose, _grid, _nearest +from tunex.cli import build_parser +from tunex.config import ( + BayesianConfig, Config, EngineConfig, MatchConfig, SearchConfig, + SprtConfig, SpsaParameter, VariantConfig, +) +from tunex.stats import WDL + + +def build_config(tmp_path: Path) -> Config: + param = SpsaParameter("SearchLMRFullMoves", 6, 4, 9, 1) + other = SpsaParameter("SearchSEEPruneMargin", 90, 60, 120, 10) + return Config( + engine=EngineConfig(path="engine"), + search=SearchConfig("movetime", 250), + match=MatchConfig(games=40, paired=True, workers=1, + results_file=str(tmp_path / "games.jsonl")), + baseline=VariantConfig("fixed", options={"Hash": 64}), + candidate=VariantConfig("candidate"), + sprt=SprtConfig(), + bayesian=BayesianConfig( + iterations=7, games_per_trial=4, startup_trials=3, + candidates=16, seed=1234, + state_file=str(tmp_path / "bayesian_state.json"), + parameters=[param, other], + ), + ) + + +class FakeRunner: + def __init__(self, cfg: Config) -> None: + self.calls = [] + + def run(self, *, baseline, candidate, games, offset): + assert baseline.options == {"Hash": 64} + assert candidate.options["Hash"] == 64 + assert games == 4 + self.calls.append((dict(candidate.options), offset)) + # deterministic, nonconstant sample objective; no real engine needed. + wins = 3 if candidate.options["SearchLMRFullMoves"] <= 6 else 1 + return WDL(wins=wins, draws=0, losses=games - wins), [] + + +def test_bayesian_config_validation(tmp_path): + cfg = build_config(tmp_path) + assert cfg.bayesian is not None + assert cfg.bayesian.games_per_trial == 4 + with pytest.raises(ValueError, match="even"): + BayesianConfig.from_dict({ + "games_per_trial": 3, + "parameters": [{"name": "SearchLMRFullMoves", "value": 6, + "min": 4, "max": 9, "step": 1}], + }) + with pytest.raises(ValueError, match="unique"): + BayesianConfig.from_dict({ + "parameters": [{"name": "X", "value": 6, "min": 4, "max": 9}, + {"name": "X", "value": 3, "min": 1, "max": 6}], + }) + + +def test_grid_and_distinct_model_candidates(tmp_path): + cfg = build_config(tmp_path).bayesian + assert cfg is not None + assert _grid(cfg.parameters[1]) == [60, 70, 80, 90, 100, 110, 120] + assert _nearest(94, cfg.parameters[1]) == 90 + rng = random.Random(11) + history = [] + for k in range(9): + proposal = _choose(rng, history, cfg) + assert tuple(proposal.items()) not in { + tuple(x["options"].items()) for x in history + } + history.append({ + "options": proposal, + "objective": 0.6 if proposal["SearchLMRFullMoves"] < 7 else 0.4, + }) + assert len({tuple(sorted(h["options"].items())) for h in history}) == 9 + + +def test_bayesian_tunes_and_resumes_deterministically(tmp_path, monkeypatch): + from tunex import bayesian + monkeypatch.setattr(bayesian, "MatchRunner", FakeRunner) + config = build_config(tmp_path) + first = BayesianRunner(config) + first.run(iterations=3) + assert len(first.trials) == 3 + assert first.trials[0]["options"] == { + "SearchLMRFullMoves": 6, "SearchSEEPruneMargin": 90, + } + stored = json.loads(Path(config.bayesian.state_file).read_text()) + assert stored["version"] == 1 + assert len(stored["trials"]) == 3 + + resumed = BayesianRunner(config) + resumed.run(iterations=7) + assert len(resumed.trials) == 7 + assert len({tuple(sorted(t["options"].items())) for t in resumed.trials}) == 7 + assert resumed.match_runner.calls[0][1] == 12 # correct game offset + assert set(resumed.best_options()) == { + "SearchLMRFullMoves", "SearchSEEPruneMargin", + } + + uninterrupted_config = build_config(tmp_path / "other") + full = BayesianRunner(uninterrupted_config) + full.run(iterations=7) + assert [t["options"] for t in resumed.trials] == [ + t["options"] for t in full.trials + ] + + +def test_resume_rejects_mismatched_experiment(tmp_path, monkeypatch): + from tunex import bayesian + monkeypatch.setattr(bayesian, "MatchRunner", FakeRunner) + config = build_config(tmp_path) + BayesianRunner(config).run(iterations=1) + config.search.value = 400 + with pytest.raises(ValueError, match="different configuration"): + BayesianRunner(config) + + +def test_space_exhaustion(tmp_path): + cfg = BayesianConfig( + iterations=10, games_per_trial=4, startup_trials=2, + parameters=[SpsaParameter("X", 0, 0, 1, 1)], + ) + previous = [ + {"options": {"X": 0}, "objective": 0.5}, + {"options": {"X": 1}, "objective": 0.5}, + ] + with pytest.raises(ValueError, match="exhausted"): + _choose(random.Random(1), previous, cfg) + + +def test_cli_bayesian_command(): + args = build_parser().parse_args(["-c", "test.json", "bayesian", "--iterations", "25"]) + assert args.command == "bayesian" + assert args.iterations == 25 + assert build_parser().parse_args(["sprt"]).command == "sprt" + assert build_parser().parse_args(["spsa"]).command == "spsa"