summaryrefslogtreecommitdiff
path: root/annotate_pgns/fit_k_and_label.py
diff options
context:
space:
mode:
authorScott Gasch <[email protected]>2026-10-15 13:11:22 -0700
committerScott Gasch <[email protected]>2026-10-15 13:11:22 -0700
commit993b6eeb2a9ac7b659c473b6883fa53af1ad3a0b (patch)
treec5388824dec15a7691ce9952a6eb4af66e837127 /annotate_pgns/fit_k_and_label.py
parent53b380c658f828c4c0e096be2763e6b012c23abd (diff)
Phase A NNUE: feature-indexing spec, annotation/training pipeline, dump nnuefeatures
Builds out the NNUE Phase A prototype per docs/NNUE.md: - src/nnuefeatures.c: standalone HalfKAv2_hm feature-index reference implementation (new `dump nnuefeatures` command), sourced from Stockfish's half_ka_v2_hm.{h,cpp} (sf_16 tag), including the 0x88->64 square conversion typhoon's own board representation needs. - annotate_pgns/: sharded TWIC/self-play annotation harness (annotate.py), position extraction, quiet-position filtering (flat and game-shaped, plus an upstream --prefilter-quiet option), train/holdout game-level stratified split, WDL-sigmoid K fitting, the Python HalfKAv2_hm reference encoder (nnue_feature_spec.py) and its bit-for-bit cross-checker against the C side (cross_check_features.py), and a first PyTorch training pipeline (nnue_model.py, nnue_train.py, build_training_tensors.py) -- a 256x2-32-32-1-ish net trained locally on CPU against ~1.85M quiet TWIC positions as a pipeline/framework validation pass. - Finish an in-progress InitMagic/InitLMRTable/InitEval -> InitializeBitboardMagic/InitializeLMRTable/InitializeEval rename: chess.h's prototypes hadn't been updated to match the already-renamed definitions/call sites, leaving 4 implicit-function-declaration warnings. - tests/arasan.epd: new curated EPD suite. - docs/: add the project's design docs (previously untracked). Co-Authored-By: Claude Sonnet 5 <[email protected]>
Diffstat (limited to 'annotate_pgns/fit_k_and_label.py')
-rw-r--r--annotate_pgns/fit_k_and_label.py146
1 files changed, 146 insertions, 0 deletions
diff --git a/annotate_pgns/fit_k_and_label.py b/annotate_pgns/fit_k_and_label.py
new file mode 100644
index 0000000..9fd09b0
--- /dev/null
+++ b/annotate_pgns/fit_k_and_label.py
@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Fit the WDL-sigmoid scale K on the TRAIN split only, then label
+every non-exact position in both train and holdout with:
+ - p_search: sigmoid(score / K) -- the search-score-derived win
+ probability estimate.
+ - r_outcome: that position's own game's actual final result,
+ converted to side-to-move-relative (matching the score's own
+ convention: 1.0 = the side to move at this FEN eventually won
+ their game, 0.0 = eventually lost, 0.5 = drawn).
+
+Does NOT blend p_search/r_outcome into a single training target --
+that mixing weight (lambda) is left as a free parameter to sweep
+later; storing both components separately means that sweep never
+needs to re-derive anything from here.
+
+K is fit by ternary search minimizing MSE(sigmoid(score/K), r_outcome)
+over every non-exact TRAIN position (not holdout -- fitting K on
+holdout data would leak information into what's supposed to be an
+untouched generalization check). The same fitted K is then applied
+(never refit) to label holdout positions too.
+
+Usage:
+ python3 fit_k_and_label.py --train runs/sd10_twic_2M/games.train.jsonl \
+ --holdout runs/sd10_twic_2M/games.holdout.jsonl \
+ --train-out runs/sd10_twic_2M/games.train.labeled.jsonl \
+ --holdout-out runs/sd10_twic_2M/games.holdout.labeled.jsonl
+"""
+import argparse
+import json
+import math
+
+
+def result_to_white_score(result):
+ """1.0/0.5/0.0 from White's perspective, or None if unknown ('*')."""
+ return {"1-0": 1.0, "0-1": 0.0, "1/2-1/2": 0.5}.get(result)
+
+
+def stm_relative_outcome(white_score, fen):
+ """Flip a White-perspective game result to the side-to-move's own
+ perspective at this FEN, matching the score field's own
+ side-to-move-relative convention."""
+ stm_is_white = fen.split()[1] == "w"
+ return white_score if stm_is_white else (1.0 - white_score)
+
+
+def collect_calibration_pairs(games_path):
+ """(score, r_outcome) for every non-exact position across all
+ games in this file."""
+ pairs = []
+ n_unknown_result = 0
+ with open(games_path) as f:
+ for line in f:
+ g = json.loads(line)
+ white_score = result_to_white_score(g["result"])
+ if white_score is None:
+ n_unknown_result += 1
+ continue
+ for pos in g["positions"]:
+ if pos.get("exact"):
+ continue
+ r = stm_relative_outcome(white_score, pos["fen"])
+ pairs.append((pos["score"], r))
+ return pairs, n_unknown_result
+
+
+def mse(pairs, k):
+ err = 0.0
+ for score, r in pairs:
+ p = 1.0 / (1.0 + math.exp(-score / k))
+ err += (r - p) ** 2
+ return err / max(len(pairs), 1)
+
+
+def fit_k(pairs, lo=0.05, hi=20.0, iterations=40):
+ for _ in range(iterations):
+ m1 = lo + (hi - lo) / 3
+ m2 = hi - (hi - lo) / 3
+ if mse(pairs, m1) < mse(pairs, m2):
+ hi = m2
+ else:
+ lo = m1
+ return (lo + hi) / 2
+
+
+def label_file(in_path, out_path, k):
+ n_games = 0
+ n_positions = 0
+ n_exact_skipped = 0
+ n_unknown_result = 0
+ with open(in_path) as fin, open(out_path, "w") as fout:
+ for line in fin:
+ g = json.loads(line)
+ white_score = result_to_white_score(g["result"])
+ if white_score is None:
+ n_unknown_result += 1
+ fout.write(json.dumps(g) + "\n")
+ continue
+ for pos in g["positions"]:
+ n_positions += 1
+ if pos.get("exact"):
+ n_exact_skipped += 1
+ continue
+ r = stm_relative_outcome(white_score, pos["fen"])
+ p_search = 1.0 / (1.0 + math.exp(-pos["score"] / k))
+ pos["p_search"] = p_search
+ pos["r_outcome"] = r
+ fout.write(json.dumps(g) + "\n")
+ n_games += 1
+ return n_games, n_positions, n_exact_skipped, n_unknown_result
+
+
+def main():
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--train", required=True)
+ ap.add_argument("--holdout", required=True)
+ ap.add_argument("--train-out", required=True)
+ ap.add_argument("--holdout-out", required=True)
+ args = ap.parse_args()
+
+ print(f"Collecting calibration pairs from {args.train} (TRAIN only) ...")
+ pairs, n_unknown = collect_calibration_pairs(args.train)
+ print(f" {len(pairs)} (score, outcome) pairs "
+ f"({n_unknown} games skipped: unknown result)")
+
+ print("Fitting K (ternary search, minimizing sigmoid-MSE against "
+ "actual game outcomes) ...")
+ k = fit_k(pairs)
+ final_mse = mse(pairs, k)
+ print(f" K = {k:.2f} (MSE at fit = {final_mse:.5f})")
+
+ print(f"\nLabeling {args.train} -> {args.train_out} ...")
+ n_g, n_p, n_exact, n_unk = label_file(args.train, args.train_out, k)
+ print(f" {n_g} games, {n_p} positions, {n_exact} exact (unlabeled), "
+ f"{n_unk} unknown-result games (unlabeled)")
+
+ print(f"Labeling {args.holdout} -> {args.holdout_out} (same K, not refit) ...")
+ n_g, n_p, n_exact, n_unk = label_file(args.holdout, args.holdout_out, k)
+ print(f" {n_g} games, {n_p} positions, {n_exact} exact (unlabeled), "
+ f"{n_unk} unknown-result games (unlabeled)")
+
+ print(f"\nFitted K = {k:.4f} -- record this; applying a different K "
+ f"later (e.g. after labeling more data) is not the same model.")
+
+
+if __name__ == "__main__":
+ main()