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
|
#!/usr/bin/env python3
"""Turn a match_controls.py-matched king-safety CSV (mate/control pairs,
fen retained -- see extract_king_features.py's 2026-09-15 fen-retention
fix) into a Texel-ready (fen, result) pool: runs the same qsearch-
agreement quiet filter used for the generic TWIC pool (this data was
NEVER filtered for quietness at mining time by design -- see mine_king_
mates.py's is_quiet_position, added 2026-09-15 -- so this step matters
here at least as much as it does for TWIC, arguably more, since "N plies
before a real mate" specifically targets the moment before a real attack
lands, exactly where a quiet-by-SEE-but-not-by-search position is most
likely).
Result-label framing (a real modeling choice, not a literal game
result): label==1 rows are "this color's king got mated in this game
from here" -- treated as a loss (0.0) for that color. label==0 rows are
the matched sound/control side -- treated as a win (1.0) for that color.
Output is always from WHITE's POV (matching eval_white_pov_batch's own
convention), so a row is flipped if mated_color_or_sampled_color is
black. This is an approximation (these aren't actual final game
results, just "survived this specific king-safety episode or not") --
flagged explicitly rather than silently treated as equivalent to a real
Texel outcome label.
"""
import argparse
import csv
import sys
from tune_eval_dna import Engine, filter_by_qsearch_agreement
def main():
ap = argparse.ArgumentParser()
ap.add_argument("matched_csv")
ap.add_argument("out_tsv")
ap.add_argument("--engine", default="../typhoon")
ap.add_argument("--qsearch-max-diff", type=int, default=30)
ap.add_argument("--dna-scratch-dir", default="/tmp/typhoon_curated_pool")
args = ap.parse_args()
with open(args.matched_csv, newline="") as f:
rows = list(csv.DictReader(f))
print(f"Loaded {len(rows)} matched rows from {args.matched_csv}",
file=sys.stderr)
fen_to_row = {}
for r in rows:
fen_to_row[r["fen"]] = r
if len(fen_to_row) != len(rows):
print(f"WARNING: {len(rows) - len(fen_to_row)} duplicate FENs "
f"collapsed -- matched.csv shouldn't have dupes normally, "
f"investigate if this is large.", file=sys.stderr)
# (fen, result) pairs for the quiet-filter -- result value here is a
# placeholder (unused by filter_by_qsearch_agreement, which only
# reads/returns fens), real result labels are attached after.
placeholder_positions = [(fen, 0.0) for fen in fen_to_row]
with Engine(args.engine, args.dna_scratch_dir) as engine:
kept = filter_by_qsearch_agreement(
engine, placeholder_positions, max_diff_cp=args.qsearch_max_diff
)
n_flipped = 0
out_rows = []
for fen, _ in kept:
r = fen_to_row[fen]
label = int(r["label"])
color = r["mated_color_or_sampled_color"]
# label==1 (mate) -> loss (0.0) for that color; label==0
# (control/sound) -> win (1.0) for that color.
result_for_color = 0.0 if label == 1 else 1.0
if color == "black":
result_white_pov = 1.0 - result_for_color
n_flipped += 1
else:
result_white_pov = result_for_color
out_rows.append((fen, result_white_pov))
with open(args.out_tsv, "w") as f:
for fen, result in out_rows:
f.write(f"{fen}\t{result}\n")
print(f"Quiet-filter kept {len(kept)}/{len(fen_to_row)} "
f"({len(kept)/len(fen_to_row):.1%})", file=sys.stderr)
print(f"Wrote {len(out_rows)} (fen, result) rows to {args.out_tsv} "
f"({n_flipped} POV-flipped from black)", file=sys.stderr)
if __name__ == "__main__":
main()
|
