diff --git a/README.md b/README.md index 47b775e..d1680f9 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,58 @@ 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. +## Texel tuning (HCE) + +Texel tuning fits **handcrafted evaluation parameters to positions with known +game outcomes**, rather than spending games on every candidate weight vector. +TuneX uses a logistic outcome prediction from the engine's **static** evaluation +and minimizes mean squared prediction error with a bounded integer coordinate +descent. This is a slow but dependency-free black-box Texel-loss optimizer; +it is **not** the analytical-gradient variant. + +**Requires:** an IronPhoenix build with the companion `eval` UCI command +returning `info string eval cp N` (a *side-to-move team* score). TuneX turns +this into **Red+Yellow team perspective** before fitting. Use handcrafted +evaluation (`UseNNUE=false`); weight tuning must not accidentally fit NNUE +output. SpellWarden currently has no runtime `Eval*` UCI options, so Texel +HCE tuning is intentionally unsupported there until the evaluator exposes them. + +Create a JSONL dataset, one object per line: + +~~~json +{"fen":"","result":1.0,"game_id":"match-001"} +{"fen":"","result":0.0,"game_id":"match-002"} +{"fen":"","result":0.5,"game_id":"match-003"} +~~~ + +`result` is **1 for Red+Yellow winning, 0 for Blue+Green winning, +0.5 for draw**, regardless of the FEN's side to move. `game_id` is required +to partition entire games into disjoint training and validation sets, preventing +later positions from the same game appearing in both splits. Only use completed, +reliable game outcomes. Do not treat the engine's current scores as labels. +Prefer many varied, nonterminal positions and a representative held-out dataset. +`fen` must be a native FEN4 snapshot (not just `4PC`). + +Copy `configs/texel.example.json`, set `engine.path`, provide a real +`configs/texel_positions.example.jsonl` file with your labeled positions, +then run: + +~~~powershell +tunex -c configs/texel.example.json texel +tunex -c configs/texel.example.json texel --epochs 12 +~~~ + +The `texel` section configures `dataset`, `epochs`, logistic scaling `k`, +`validation_fraction`, `seed`, `state_file`, and integer `parameters` +with `name`, `value`, `min`, `max`, `step`. Each epoch considers +neighbouring integer values per weight, accepting changes that reduce training +loss. TuneX saves current weights and the weights with **lowest validation +loss**, and resumes from a compatible state file. It rejects incompatible +configurations, datasets, and malformed labels. + +Texel loss is not Elo: compare the selected HCE weights against the original +settings in fresh **paired SPRT matches** before changing production defaults. + ## Requirements - Python **3.11+** diff --git a/configs/texel.example.json b/configs/texel.example.json new file mode 100644 index 0000000..d1d3b95 --- /dev/null +++ b/configs/texel.example.json @@ -0,0 +1,73 @@ +{ + "variant": "four_player_teams", + "engine": { + "path": "C:/path/to/IronPhoenix/out/build/x64-release/ironphoenix.exe", + "startup_timeout": 15, + "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/games.jsonl" + }, + "baseline": { + "name": "Fixed HCE context", + "options": {} + }, + "candidate": { + "name": "Texel candidate", + "options": {} + }, + "texel": { + "dataset": "texel_positions.example.jsonl", + "epochs": 8, + "k": 1, + "validation_fraction": 0.2, + "seed": 20261113, + "state_file": "results/texel_state.json", + "parameters": [ + { + "name": "EvalKnightMobility", + "value": 9, + "min": 0, + "max": 24, + "step": 1 + }, + { + "name": "EvalBishopPairBonus", + "value": 62, + "min": 20, + "max": 100, + "step": 2 + }, + { + "name": "EvalKingZoneAttackPenalty", + "value": 77, + "min": 30, + "max": 140, + "step": 3 + }, + { + "name": "EvalKnightOutpostBonus", + "value": 14, + "min": 0, + "max": 40, + "step": 2 + } + ] + } +} diff --git a/src/tunex/cli.py b/src/tunex/cli.py index 4bb2668..2584690 100644 --- a/src/tunex/cli.py +++ b/src/tunex/cli.py @@ -11,6 +11,7 @@ from .match import GameRecord, GameSpec, MatchRunner from .sprt import SprtState from .spsa import SpsaRunner +from .texel import TexelRunner from .stats import WDL, format_summary @@ -178,6 +179,15 @@ def command_bayesian(config: Config, iterations: int | None) -> int: return 0 +def command_texel(config: Config, epochs: int | None) -> int: + runner = TexelRunner(config) + runner.run(epochs) + print("best validation HCE weights (verify in independent matches):") + for name, value in runner.best_weights.items(): + print(f" {name} = {value}") + return 0 + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="tunex", @@ -204,6 +214,9 @@ def build_parser() -> argparse.ArgumentParser: bayesian = sub.add_parser("bayesian", help="run resumable Bayesian TPE tuning") bayesian.add_argument("--iterations", type=int, default=None) + texel = sub.add_parser("texel", help="fit HCE weights to labeled positions") + texel.add_argument("--epochs", type=int, default=None) + return parser @@ -223,6 +236,8 @@ def main(argv: list[str] | None = None) -> int: return command_spsa(config, args.iterations) if args.command == "bayesian": return command_bayesian(config, args.iterations) + if args.command == "texel": + return command_texel(config, args.epochs) 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 2bd05d6..dea937d 100644 --- a/src/tunex/config.py +++ b/src/tunex/config.py @@ -251,6 +251,47 @@ def from_dict(cls, data: dict[str, Any]) -> "BayesianConfig": return cfg +@dataclass(slots=True) +class TexelConfig: + """Labeled-position Texel fitting via bounded integer coordinate descent.""" + + dataset: str + parameters: list[SpsaParameter] + epochs: int = 5 + k: float = 1.0 + validation_fraction: float = 0.2 + seed: int = 20261113 + state_file: str = "results/texel_state.json" + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "TexelConfig": + params = [SpsaParameter.from_dict(p) for p in data.get("parameters", [])] + cfg = cls( + dataset=str(data.get("dataset", "")), + parameters=params, + epochs=int(data.get("epochs", 5)), + k=float(data.get("k", 1.0)), + validation_fraction=float(data.get("validation_fraction", 0.2)), + seed=int(data.get("seed", 20261113)), + state_file=str(data.get("state_file", "results/texel_state.json")), + ) + if not cfg.dataset.strip() or not params: + raise ValueError("texel dataset and parameters are required") + if cfg.epochs < 1 or not 0 < cfg.k <= 10: + raise ValueError("texel epochs must be positive and k in (0, 10]") + if not 0 < cfg.validation_fraction < 0.5: + raise ValueError("texel validation_fraction must be in (0, 0.5)") + if len({p.name for p in params}) != len(params): + raise ValueError("texel parameter names must be unique") + for p in params: + if any(abs(v - round(v)) > 1e-9 for v in + (p.minimum, p.maximum, p.value, p.step)): + raise ValueError(f"{p.name}: texel requires integer-valued parameters") + if p.minimum == p.maximum or p.step < 1: + raise ValueError(f"{p.name}: texel bounds must vary with step >= 1") + return cfg + + @dataclass(slots=True) class Config: engine: EngineConfig @@ -261,6 +302,7 @@ class Config: sprt: SprtConfig spsa: SpsaConfig | None = None bayesian: BayesianConfig | None = None + texel: TexelConfig | None = None variant: str = "four_player_teams" @classmethod @@ -270,6 +312,10 @@ def load(cls, path: str | Path) -> "Config": variant = str(data.get("variant", "four_player_teams")).lower() if variant not in {"four_player_teams", "spell_chess"}: raise ValueError("variant must be four_player_teams or spell_chess") + if "texel" in data and data["texel"].get("dataset"): + dataset = Path(data["texel"]["dataset"]) + if not dataset.is_absolute(): + data["texel"]["dataset"] = str((config_path.parent / dataset).resolve()) engine = EngineConfig.from_dict(data.get("engine", {})) match = MatchConfig.from_dict(data.get("match")) openings = match.openings_file @@ -292,5 +338,6 @@ def load(cls, path: str | Path) -> "Config": spsa=SpsaConfig.from_dict(data["spsa"]) if "spsa" in data else None, bayesian=BayesianConfig.from_dict(data["bayesian"]) if "bayesian" in data else None, + texel=TexelConfig.from_dict(data["texel"]) if "texel" in data else None, variant=variant, ) diff --git a/src/tunex/engine.py b/src/tunex/engine.py index 4ef1ebe..1559c73 100644 --- a/src/tunex/engine.py +++ b/src/tunex/engine.py @@ -220,6 +220,33 @@ def search(self, position_command: str, go_command: str) -> SearchResult: info_line=last_info, ) + def static_eval_cp(self, fen: str) -> int: + """Read the engine's *static*, side-to-move evaluation (not a search score). + + IronPhoenix and SpellWarden use 'info string eval cp N' in response + to 'eval'. Reject missing or malformed responses, including bad FENs. + """ + self.send(f"position fen {fen}") + self.send("eval") + self.send("isready") + lines = self._read_until(lambda line: line == "readyok", self.startup_timeout) + for line in lines: + if line.startswith("info string ") and any(term in line.lower() for term in ( + "invalid", "illegal", "error", "failed", "requires", "refused", + )): + raise EngineError(f"{self.name}: {line}") + scores = [ + re.fullmatch(r"info string eval cp (-?\d+)", line) + for line in lines + if line.startswith("info string eval cp ") + ] + if len(scores) != 1 or scores[0] is None: + raise EngineError( + f"{self.name} must support 'eval' -> 'info string eval cp N' " + "for Texel tuning" + ) + return int(scores[0].group(1)) + def opening_position(self, position_command: str) -> tuple[str, tuple[str, ...]]: """Get native FEN4 + legal UCI moves for SAN opening conversion. diff --git a/src/tunex/texel.py b/src/tunex/texel.py new file mode 100644 index 0000000..8a9edbb --- /dev/null +++ b/src/tunex/texel.py @@ -0,0 +1,207 @@ +"""Texel tuning of HCE parameters from outcome-labeled positions. + +Uses an engine's true static evaluation rather than search scores. The +objective is mean squared error of sigmoid(K * cp / 400) against 0/0.5/1 +outcomes from a fixed team/colour perspective. Integer coordinate descent +is a dependency-free baseline for expensive black-box eval callbacks. +""" +from __future__ import annotations + +from dataclasses import asdict, dataclass +import hashlib +import json +import math +import os +from pathlib import Path +import random + +from .config import Config, SpsaParameter +from .engine import UciEngine + + +@dataclass(frozen=True) +class Sample: + fen: str + result: float + game: str + + +def read_dataset(path: str, variant: str) -> list[Sample]: + samples: list[Sample] = [] + for number, raw in enumerate(Path(path).read_text(encoding="utf-8-sig").splitlines(), 1): + if not raw.strip(): + continue + try: + entry = json.loads(raw) + fen = entry["fen"] + result = entry["result"] + game = entry["game_id"] + except (ValueError, KeyError, TypeError) as exc: + raise ValueError(f"dataset line {number}: expected fen/result/game_id JSON") from exc + if not isinstance(fen, str) or not fen.strip() or not isinstance(game, (str, int)): + raise ValueError(f"dataset line {number}: invalid fen or game_id") + if type(result) not in (float, int) or float(result) not in (0, 0.5, 1): + raise ValueError(f"dataset line {number}: result must be 0, 0.5 or 1") + parts = fen.split() + if variant == "four_player_teams": + if fen[0].upper() not in "RBYG" or fen[1:2] != "-": + raise ValueError(f"dataset line {number}: expected native FEN4") + elif len(parts) < 6 or parts[1] not in ("w", "b"): + raise ValueError(f"dataset line {number}: expected standard 8x8 FEN") + samples.append(Sample(fen, float(result), str(game))) + if len(samples) < 2 or len({p.game for p in samples}) < 2: + raise ValueError("Texel tuning needs positions from at least two distinct games") + return samples + + +def split_games(samples: list[Sample], seed: int, fraction: float) -> tuple[list[Sample], list[Sample]]: + groups = sorted({p.game for p in samples}) + random.Random(seed).shuffle(groups) + holdout = set(groups[:max(1, min(len(groups) - 1, round(len(groups) * fraction)))]) + return ([p for p in samples if p.game not in holdout], + [p for p in samples if p.game in holdout]) + + +def prediction(cp: int, k: float) -> float: + # Avoid overflow in logistic tails. + z = max(-60., min(60., k * cp / 400.0 * math.log(10))) + return 1.0 / (1.0 + math.exp(-z)) + + +def sign(fen: str, variant: str) -> int: + # Static eval from both engines is side-to-move relative. + # Labels are always from R+Y (4PC) or White (SpellWarden). + if variant == "four_player_teams": + return 1 if fen[0].upper() in ("R", "Y") else -1 + return 1 if fen.split()[1] == "w" else -1 + + +def grid_value(value: float, p: SpsaParameter) -> int: + low, high, step = int(p.minimum), int(p.maximum), int(p.step) + n = max(0, min((high - low) // step, round((value - low) / step))) + return low + n * step + + +def signature(config: Config, digest: str) -> str: + tune = config.texel + assert tune is not None + return json.dumps({ + "engine": asdict(config.engine), "baseline": asdict(config.baseline), + "variant": config.variant, "texel": asdict(tune), "dataset_sha256": digest, + }, sort_keys=True) + + +class TexelRunner: + def __init__(self, config: Config) -> None: + if config.texel is None: + raise ValueError("config has no texel section") + self.config = config + self.tune = config.texel + if config.variant == "spell_chess": + # SpellWarden's evaluation weights are presently hard-coded and + # not UCI-tunable. Refuse to pretend these are trainable. + raise ValueError("SpellWarden has no Eval* UCI HCE options yet") + data = Path(self.tune.dataset).read_bytes() + self.digest = hashlib.sha256(data).hexdigest() + self.training, self.validation = split_games( + read_dataset(self.tune.dataset, config.variant), + self.tune.seed, self.tune.validation_fraction, + ) + self.weights = {p.name: grid_value(p.value, p) for p in self.tune.parameters} + self.best_weights = dict(self.weights) + self.best_validation = math.inf + self.epochs_done = 0 + self._load() + + @property + def state_path(self) -> Path: + return Path(self.tune.state_file) + + def _load(self) -> None: + if not self.state_path.exists(): + return + doc = json.loads(self.state_path.read_text(encoding="utf-8")) + if doc.get("signature") != signature(self.config, self.digest): + raise ValueError("Texel resume state does not match config/dataset") + self.weights = doc["weights"] + self.best_weights = doc["best_weights"] + self.best_validation = float(doc["best_validation"]) + self.epochs_done = int(doc["epochs_done"]) + + def _save(self) -> None: + path = self.state_path + path.parent.mkdir(parents=True, exist_ok=True) + doc = { + "signature": signature(self.config, self.digest), + "weights": self.weights, "best_weights": self.best_weights, + "best_validation": self.best_validation, "epochs_done": self.epochs_done, + } + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(json.dumps(doc, indent=2), encoding="utf-8") + os.replace(tmp, path) + + def run(self, epochs: int | None = None) -> None: + target = self.tune.epochs if epochs is None else epochs + if target < 1: + raise ValueError("Texel epochs must be positive") + path = self.config.baseline.path or self.config.engine.path + with UciEngine( + path, name="Texel HCE", startup_timeout=self.config.engine.startup_timeout, + move_timeout=self.config.engine.move_timeout, + strict_options=True, + ) as engine: + fixed = dict(self.config.engine.common_options) + fixed.update(self.config.baseline.options) + # Never fit an NNUE prediction as though it were HCE. + fixed["UseNNUE"] = False + engine.configure(fixed) + for p in self.tune.parameters: + if p.name not in engine.options or not p.name.startswith("Eval"): + raise ValueError(f"unsupported HCE UCI parameter {p.name!r}") + engine.new_game() + + def mse(weights: dict[str, int], samples: list[Sample]) -> float: + engine.configure(weights) + errors = [] + for sample in samples: + cp = engine.static_eval_cp(sample.fen) * sign( + sample.fen, self.config.variant + ) + score = prediction(cp, self.tune.k) + errors.append((score - sample.result) ** 2) + return sum(errors) / len(errors) + + for epoch in range(self.epochs_done, target): + # Coordinate descent with shrinking step radius, bounded to + # each parameter's grid. Consider both directions; never + # accept a worsening training objective. + base_loss = mse(self.weights, self.training) + for p in self.tune.parameters: + current = self.weights[p.name] + stride = max(1, round((target - epoch) / max(1, target) * 4)) + neighbors = { + grid_value(current + delta * stride * p.step, p) + for delta in (-1, 1) + } + for value in sorted(neighbors): + if value == current: + continue + candidate = dict(self.weights) + candidate[p.name] = value + loss = mse(candidate, self.training) + if loss < base_loss - 1e-12: + self.weights = candidate + current = value + base_loss = loss + validation = mse(self.weights, self.validation) + if validation < self.best_validation: + self.best_validation = validation + self.best_weights = dict(self.weights) + self.epochs_done = epoch + 1 + self._save() + print( + f"[Texel {self.epochs_done}/{target}] train MSE {base_loss:.6f} " + f"validation MSE {validation:.6f} best {self.best_validation:.6f}", + flush=True, + ) + print("Best validation weights:", self.best_weights) diff --git a/tests/test_texel.py b/tests/test_texel.py new file mode 100644 index 0000000..8d9a28c --- /dev/null +++ b/tests/test_texel.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import patch + +import pytest + +from tunex.config import ( + Config, EngineConfig, SearchConfig, MatchConfig, SprtConfig, + VariantConfig, TexelConfig, SpsaParameter, +) +from tunex.texel import ( + TexelRunner, read_dataset, split_games, prediction, sign, grid_value, +) +from tunex.cli import build_parser + + +FEN_R = "R-0,0,0,0-1,1,1,1-1,1,1,1-0,0,0,0-0-{}-x,x,x,yK,1,yR,x,x,x/x,x,x,8,x,x,x" +FEN_B = FEN_R.replace("R-", "B-", 1) + + +def make_data(path: Path) -> None: + records = [] + for game in range(10): + for fen in (FEN_R, FEN_B): + records.append({"fen": fen, "result": float(game % 2), + "game_id": f"game-{game}"}) + path.write_text("\n".join(map(json.dumps, records)) + "\n", encoding="utf-8") + + +def cfg(tmp_path: Path) -> Config: + data = tmp_path / "data.jsonl" + make_data(data) + return Config( + engine=EngineConfig("test-engine", common_options={"UseNNUE": False}), + search=SearchConfig(), + match=MatchConfig(), + baseline=VariantConfig("original"), + candidate=VariantConfig("candidate"), + sprt=SprtConfig(), + texel=TexelConfig( + dataset=str(data), parameters=[ + SpsaParameter("EvalKnightMobility", 9, 0, 20, 1) + ], + epochs=2, validation_fraction=0.2, + state_file=str(tmp_path / "texel.json"), + ), + ) + + +class FakeUci: + def __init__(self, *args, **kwargs): + self.options = {"EvalKnightMobility"} + self.current = 9 + self.closed = False + + def __enter__(self): + return self + + def __exit__(self, *args): + self.closed = True + + def configure(self, options): + self.current = options.get("EvalKnightMobility", self.current) + + def new_game(self): + pass + + def static_eval_cp(self, fen): + # Synthetic black-box HCE, not search scores. + return (self.current * 20) * (1 if fen.startswith("R-") else -1) + + +def test_labels_and_perspective(tmp_path): + file = tmp_path / "data.jsonl" + make_data(file) + samples = read_dataset(str(file), "four_player_teams") + train, valid = split_games(samples, 123, 0.2) + assert set(s.game for s in train).isdisjoint(s.game for s in valid) + assert len(train) + len(valid) == 20 + assert sign(FEN_R, "four_player_teams") == 1 + assert sign(FEN_B, "four_player_teams") == -1 + assert sign("4k3/8/8/8/8/8/8/4K3 b - - 0 1", "spell_chess") == -1 + assert prediction(0, 1.0) == 0.5 + assert prediction(10000, 1.0) > 0.9999 + assert grid_value(12.9, SpsaParameter("x", 10, 0, 20, 3)) == 12 + + +def test_invalid_data(tmp_path): + file = tmp_path / "data.jsonl" + file.write_text(json.dumps({"fen": FEN_R, "result": 2, "game_id": "x"}) + "\n") + with pytest.raises(ValueError, match="result must"): + read_dataset(str(file), "four_player_teams") + file.write_text(json.dumps({"fen": FEN_R, "result": 1}) + "\n") + with pytest.raises(ValueError, match="fen/result/game_id"): + read_dataset(str(file), "four_player_teams") + + +def test_fitting_and_resume(tmp_path): + config = cfg(tmp_path) + with patch("tunex.texel.UciEngine", FakeUci): + runner = TexelRunner(config) + runner.run(1) + assert runner.epochs_done == 1 + assert runner.best_validation < float("inf") + resumed = TexelRunner(config) + resumed.run(2) + assert resumed.epochs_done == 2 + assert Path(config.texel.state_file).exists() + + # Only best validation weights are presented for a held-out test. + assert "EvalKnightMobility" in resumed.best_weights + config.texel.k = 2 + with pytest.raises(ValueError, match="does not match"): + TexelRunner(config) + + +def test_config_validation_and_cli(tmp_path): + data = cfg(tmp_path) + assert data.texel is not None + assert build_parser().parse_args(["texel", "--epochs", "4"]).epochs == 4 + with pytest.raises(ValueError, match="required"): + TexelConfig.from_dict({"dataset": ""}) + with pytest.raises(ValueError, match="integer"): + TexelConfig.from_dict({"dataset": "a", "parameters": [ + {"name": "EvalPawnValue", "value": 86.5, "min": 80, "max": 100} + ]}) + data.variant = "spell_chess" + with pytest.raises(ValueError, match="no Eval"): + TexelRunner(data)