1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
#!/usr/bin/env python3
"""Compares our trained checkpoints against a real, matching-commit
Stockfish build's NNUE evaluation on sampled real positions.
NOT a weight-reuse step (see conversation: the goal is a from-scratch
net) -- this is purely a diagnostic to see how far our net's score
scale/ranking diverges from a known-strong reference, and whether
feature factorization measurably closes that gap vs. the unfactored
run.
Stockfish's "NNUE evaluation" line (not "Final evaluation", which
blends in a classical/hybrid term we have no equivalent of) is the
directly comparable quantity -- both it and our forward() are pure
NNUE output, no classical blending, both white-side-relative after
sign conversion.
Usage:
python3 compare_vs_stockfish_net.py \
--sample-file runs/stockfish_selfplay/games.holdout.labeled.quiet.jsonl \
--n-positions 1000 \
--sf-binary /tmp/sf_38e616/src/stockfish \
--sf-net nn-fdc1d0fe6455.nnue \
--ckpt-unfactored runs/stockfish_selfplay/nnue_ckpt_lambda075/ckpt_epoch033.pt \
--ckpt-factored runs/stockfish_selfplay/nnue_ckpt_lambda075_factored/ckpt_epoch033.pt
"""
import argparse
import json
import random
import re
import subprocess
import sys
import chess
import numpy as np
import torch
from nnue_feature_spec import active_feature_indices
from nnue_model import NnueNet
NNUE_EVAL_RE = re.compile(r"NNUE evaluation\s+([+-]?\d+\.\d+) \(white side\)")
def sample_fens(sample_file, n, seed):
rng = random.Random(seed)
reservoir = []
with open(sample_file) as f:
seen = 0
for line in f:
game = json.loads(line)
for pos in game["positions"]:
if pos.get("exact"):
continue
seen += 1
if len(reservoir) < n:
reservoir.append(pos["fen"])
else:
j = rng.randint(0, seen - 1)
if j < n:
reservoir[j] = pos["fen"]
return reservoir
def load_model(ckpt_path):
ckpt = torch.load(ckpt_path, map_location="cpu", weights_only=True)
model = NnueNet()
missing, unexpected = model.load_state_dict(ckpt["model"], strict=False)
bad_missing = [k for k in missing if k != "ft_p.weight"]
if bad_missing or unexpected:
sys.exit(f"{ckpt_path}: load_state_dict mismatch -- "
f"missing={bad_missing}, unexpected={unexpected}")
model.eval()
return model
def our_score_white_relative(model, fen):
board = chess.Board(fen)
w = active_feature_indices(board, chess.WHITE)
b = active_feature_indices(board, chess.BLACK)
w_idx = torch.tensor(w, dtype=torch.long)
w_off = torch.tensor([0], dtype=torch.long)
b_idx = torch.tensor(b, dtype=torch.long)
b_off = torch.tensor([0], dtype=torch.long)
stm = torch.tensor([1 if board.turn == chess.WHITE else 0], dtype=torch.long)
with torch.no_grad():
raw = model(w_idx, w_off, b_idx, b_off, stm).item()
return raw if board.turn == chess.WHITE else -raw
def stockfish_scores_white_relative(sf_binary, sf_net, fens):
"""One persistent Stockfish process, one 'position fen ...; eval'
round-trip per FEN, parsing the 'NNUE evaluation' line. Much
cheaper than spawning a fresh process per position."""
proc = subprocess.Popen(
[sf_binary], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, text=True, bufsize=1)
def send(cmd):
proc.stdin.write(cmd + "\n")
proc.stdin.flush()
send(f"setoption name EvalFile value {sf_net}")
send("isready")
while True:
line = proc.stdout.readline()
if not line or "readyok" in line:
break
scores = []
for fen in fens:
send(f"position fen {fen}")
send("eval")
score = None
while True:
line = proc.stdout.readline()
if not line:
sys.exit("Stockfish process died mid-eval -- net likely "
"failed to load (check EvalFile path/error output)")
m = NNUE_EVAL_RE.search(line)
if m:
score = float(m.group(1))
if "Final evaluation" in line:
break
scores.append(score)
send("quit")
proc.wait(timeout=5)
return scores
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--sample-file", required=True)
ap.add_argument("--n-positions", type=int, default=1000)
ap.add_argument("--seed", type=int, default=0)
ap.add_argument("--sf-binary", required=True)
ap.add_argument("--sf-net", required=True)
ap.add_argument("--ckpt-unfactored", required=True)
ap.add_argument("--ckpt-factored", required=True)
args = ap.parse_args()
print(f"Sampling {args.n_positions} non-exact positions from {args.sample_file} ...")
fens = sample_fens(args.sample_file, args.n_positions, args.seed)
print(f" got {len(fens)} FENs")
print("Loading checkpoints ...")
model_unfactored = load_model(args.ckpt_unfactored)
model_factored = load_model(args.ckpt_factored)
print(f"Running {args.sf_binary} (net={args.sf_net}) ...")
sf_scores = stockfish_scores_white_relative(args.sf_binary, args.sf_net, fens)
print("Running our checkpoints ...")
ours_unfactored = [our_score_white_relative(model_unfactored, fen) for fen in fens]
ours_factored = [our_score_white_relative(model_factored, fen) for fen in fens]
sf = np.array(sf_scores, dtype=np.float64)
uf = np.array(ours_unfactored, dtype=np.float64)
fa = np.array(ours_factored, dtype=np.float64)
def report(name, ours):
diff = ours - sf
corr = np.corrcoef(ours, sf)[0, 1]
# best-fit scale (least squares, no intercept) -- how far off is
# our net's overall SCALE from Stockfish's, independent of the
# per-position noise captured by correlation/residual.
scale = float(np.dot(ours, sf) / np.dot(ours, ours))
resid_after_scale = ours * scale - sf
print(f"\n--- {name} vs Stockfish (n={len(sf)}) ---")
print(f" correlation: {corr:.4f}")
print(f" raw mean|diff|: {np.abs(diff).mean():.3f} pawns")
print(f" raw median diff: {np.median(diff):.3f} pawns")
print(f" best-fit scale (ours->sf): {scale:.3f}")
print(f" mean|resid| after scale-fit: {np.abs(resid_after_scale).mean():.3f} pawns")
report("Unfactored", uf)
report("Factored", fa)
np.savez("/tmp/compare_vs_sf.npz", sf=sf, unfactored=uf, factored=fa, fens=np.array(fens))
print("\nSaved raw arrays to /tmp/compare_vs_sf.npz")
if __name__ == "__main__":
sys.exit(main())
|
