diff --git a/cleaning/format.py b/cleaning/format.py deleted file mode 100644 index a914834..0000000 --- a/cleaning/format.py +++ /dev/null @@ -1,168 +0,0 @@ -#!/usr/bin/env python3 -""" -Turn refined PGN games (Site/WhiteElo/BlackElo headers + movetext with -%eval comments) into a CSV where each row is the board state *after* one -move. - -Row layout (67 columns): - game_id, a8, b8, c8, d8, e8, f8, g8, h8, a7, ... , h1, eval, turn - - - game_id : 0-based index of the game in the input file. Not asked for - explicitly, but with many games concatenated into one CSV - there is no other way to tell which rows belong to which - game, so I added it as column 1. Drop it with --no-game-id - if you don't want it. - - a8..h1 : the 64 squares (rank 8 down to rank 1, file a to h), each - holding the FEN piece letter on that square after the move - (uppercase = white, lowercase = black) or an empty string - if the square is empty. - - eval : the %eval attached to that move (blank if that particular - move had none). - - turn : "white" or "black" -- whose turn it is to move NEXT. - -Streams game-by-game (like the two earlier scripts) so it stays cheap on -RAM no matter how large the input file is. Requires `pip install chess`. - -Usage: - python pgn_to_csv.py refined.pgn out.csv - python pgn_to_csv.py refined.pgn out.csv --no-game-id -""" - -import argparse -import csv -import re -import sys - -import chess - -TAG_RE = re.compile(r'^\[(\w+)\s+"(.*)"\]\s*$') -EVAL_RE = re.compile(r'\[%eval\s+([^\]]+)\]') -MOVE_NUM_RE = re.compile(r'^\d+\.+$') -RESULT_RE = re.compile(r'^(1-0|0-1|1/2-1/2|\*)$') -# Strip trailing annotation glyphs python-chess's SAN parser won't accept -# (e.g. "e5?!", "Nc3!!", "Qxd1+!") while keeping the "+"/"#" that ARE part -# of legal SAN. -SUFFIX_ANNOTATION_RE = re.compile(r'[!?]+$') - -SQUARE_COLUMNS = [f + str(r) for r in range(8, 0, -1) for f in "abcdefgh"] - - -def tokenize_movetext(movetext: str): - """Yield (san_move, eval_str_or_None) pairs from a movetext blob.""" - tokens = re.findall(r'\{[^}]*\}|\S+', movetext) - i = 0 - n = len(tokens) - while i < n: - tok = tokens[i] - if MOVE_NUM_RE.match(tok) or RESULT_RE.match(tok): - i += 1 - continue - move_san = tok - eval_val = None - if i + 1 < n and tokens[i + 1].startswith('{'): - m = EVAL_RE.search(tokens[i + 1]) - if m: - eval_val = m.group(1) - i += 2 - else: - i += 1 - yield move_san, eval_val - - -def board_row(board: chess.Board): - """Return the 64 square values for the current board, a8..h1 order.""" - row = [] - for name in SQUARE_COLUMNS: - sq = chess.parse_square(name) - piece = board.piece_at(sq) - row.append(piece.symbol() if piece else "") - return row - - -def process_game(game_id, header_lines, movetext_lines, writer, include_game_id): - if not movetext_lines: - return 0 - - movetext = ' '.join(line.strip() for line in movetext_lines if line.strip()) - - board = chess.Board() - rows_written = 0 - - for move_san, eval_val in tokenize_movetext(movetext): - clean_san = SUFFIX_ANNOTATION_RE.sub('', move_san) - try: - board.push_san(clean_san) - except (chess.IllegalMoveError, chess.InvalidMoveError, chess.AmbiguousMoveError, ValueError) as e: - print(f" [game {game_id}] stopping early: could not parse move " - f"'{move_san}' ({e})", file=sys.stderr) - break - - turn = "white" if board.turn == chess.WHITE else "black" - row = (board_row(board) + [eval_val if eval_val is not None else "", turn]) - if include_game_id: - row = [game_id] + row - writer.writerow(row) - rows_written += 1 - - return rows_written - - -def refine(input_path: str, output_path: str, include_game_id: bool): - header_lines = [] - movetext_lines = [] - state = "seek_header" - game_id = 0 - total_rows = 0 - - header_row = SQUARE_COLUMNS + ["eval", "turn"] - if include_game_id: - header_row = ["game_id"] + header_row - - with open(input_path, "r", encoding="utf-8", errors="replace", buffering=1024 * 1024) as fin, \ - open(output_path, "w", encoding="utf-8", newline="") as fout: - - writer = csv.writer(fout) - writer.writerow(header_row) - - for raw_line in fin: - line = raw_line.rstrip("\n") - - if state == "seek_header": - if line.startswith("["): - header_lines.append(line) - elif line.strip() == "" and header_lines: - state = "seek_movetext" - elif state == "seek_movetext": - if line.strip() == "": - total_rows += process_game(game_id, header_lines, movetext_lines, - writer, include_game_id) - game_id += 1 - header_lines = [] - movetext_lines = [] - state = "seek_header" - else: - movetext_lines.append(line) - - if header_lines and movetext_lines: - total_rows += process_game(game_id, header_lines, movetext_lines, - writer, include_game_id) - game_id += 1 - - return game_id, total_rows - - -def main(): - parser = argparse.ArgumentParser( - description="Convert refined PGN games into a per-move board-state CSV.") - parser.add_argument("input", help="Path to the refined PGN file") - parser.add_argument("output", help="Path to write the CSV file") - parser.add_argument("--no-game-id", action="store_true", - help="Omit the leading game_id column (66-col strict layout).") - args = parser.parse_args() - - games, rows = refine(args.input, args.output, include_game_id=not args.no_game_id) - print(f"Processed {games} games -> {rows} rows written to {args.output}") - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2eea298 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +python-chess + +torch +pandas +pandas-stubs \ No newline at end of file diff --git a/src/data_processing/dataset_loader.py b/src/data_processing/dataset_loader.py new file mode 100644 index 0000000..d905b9b --- /dev/null +++ b/src/data_processing/dataset_loader.py @@ -0,0 +1,97 @@ +import pandas as pd +import torch + +INPUT_CSV = "../../dataset/output.csv" +OUTPUT_FILE = "../../dataset/output.pt" + +SQUARES = [ + f"{file}{rank}" + for rank in range(8, 0, -1) + for file in "abcdefgh" +] + +PIECE_TO_PLANE = { + "P": 0, + "N": 1, + "B": 2, + "R": 3, + "Q": 4, + "K": 5, + "p": 6, + "n": 7, + "b": 8, + "r": 9, + "q": 10, + "k": 11, +} + +def board_to_planes(row): + planes = torch.zeros((12, 8, 8), dtype=torch.float32) + + for index, square in enumerate(SQUARES): + piece = row[square] + + if pd.isna(piece): + continue + + plane = PIECE_TO_PLANE.get(piece) + + if plane is None: + continue + + row_index = index // 8 + col_index = index % 8 + + planes[plane, row_index, col_index] = 1.0 + + return planes + +def parse_eval(value): + value = str(value).strip() + + if value.startswith("#"): + mate_distance = int(value[1:]) + + if mate_distance > 0: + return 10.0 + else: + return -10.0 + + return float(value) + +def main(): + df = pd.read_csv(INPUT_CSV) + + positions = [] + evaluations = [] + turns = [] + + for _, row in df.iterrows(): + positions.append(board_to_planes(row)) + + evaluations.append(parse_eval(row["eval"])) + + turns.append(1 if row["turn"] == "white" else 0) + + + X = torch.stack(positions) + y = torch.tensor(evaluations, dtype=torch.float32) + turn = torch.tensor(turns, dtype=torch.float32) + + torch.save( + { + "X": X, + "y": y, + "turn": turn, + }, + OUTPUT_FILE, + ) + + print(f"Loaded {len(df):,} positions") + print(f"X shape: {X.shape}") + print(f"y shape: {y.shape}") + print(f"turn shape: {turn.shape}") + print(f"Saved to: {OUTPUT_FILE}") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/cleaning/extract.py b/src/data_processing/extract.py similarity index 100% rename from cleaning/extract.py rename to src/data_processing/extract.py diff --git a/src/data_processing/format.py b/src/data_processing/format.py new file mode 100644 index 0000000..72f4ac8 --- /dev/null +++ b/src/data_processing/format.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python3 +""" +Turn refined PGN games into a CSV where each row is the board state +after one evaluated move. + +Row layout: + game_id, a8, b8, ..., h1, eval, turn + + - game_id : Lichess game ID extracted from the Site header. + Example: + [Site "https://lichess.org/qE0J8lxe"] + becomes: + qE0J8lxe + + - a8..h1 : the 64 squares after the move. + Uppercase = white, lowercase = black, + empty string = empty square. + + - eval : the %eval attached to the move. + + - turn : "white" or "black", meaning whose turn it is NEXT. + +Moves without an evaluation are skipped. + +Streams game-by-game, so memory usage stays low. + +Requires: + pip install chess + +Usage: + python pgn_to_csv.py refined.pgn out.csv + python pgn_to_csv.py refined.pgn out.csv --no-game-id +""" + +import argparse +import csv +import re +import sys + +import chess + +TAG_RE = re.compile(r'^\[(\w+)\s+"(.*)"\]\s*$') +EVAL_RE = re.compile(r'\[%eval\s+([^\]]+)\]') +MOVE_NUM_RE = re.compile(r'^\d+\.+$') +RESULT_RE = re.compile(r'^(1-0|0-1|1/2-1/2|\*)$') +SUFFIX_ANNOTATION_RE = re.compile(r'[!?]+$') + +SQUARE_COLUMNS = [ + f + str(r) + for r in range(8, 0, -1) + for f in "abcdefgh" +] + + +def parse_headers(header_lines): + """ + Convert PGN header lines into a dictionary. + + Example: + [Site "https://lichess.org/qE0J8lxe"] + + becomes: + { + "Site": "https://lichess.org/qE0J8lxe" + } + """ + headers = {} + + for line in header_lines: + match = TAG_RE.match(line) + + if match: + key, value = match.groups() + headers[key] = value + + return headers + + +def extract_game_id(site): + """ + Extract the Lichess game ID from the Site URL. + + Example: + https://lichess.org/qE0J8lxe + -> qE0J8lxe + """ + if not site: + return None + + return site.rstrip("/").split("/")[-1] + + +def tokenize_movetext(movetext): + """ + Yield (san_move, eval_str_or_None) pairs from a movetext blob. + """ + tokens = re.findall(r'\{[^}]*\}|\S+', movetext) + + i = 0 + n = len(tokens) + + while i < n: + tok = tokens[i] + + if MOVE_NUM_RE.match(tok) or RESULT_RE.match(tok): + i += 1 + continue + + move_san = tok + eval_val = None + + if i + 1 < n and tokens[i + 1].startswith('{'): + match = EVAL_RE.search(tokens[i + 1]) + + if match: + eval_val = match.group(1) + + i += 2 + + else: + i += 1 + + yield move_san, eval_val + + +def board_row(board): + """ + Return the 64 square values for the current board, + in a8..h1 order. + """ + row = [] + + for name in SQUARE_COLUMNS: + square = chess.parse_square(name) + piece = board.piece_at(square) + + row.append(piece.symbol() if piece else "") + + return row + + +def process_game(header_lines, movetext_lines, writer, include_game_id): + """ + Process one game and write its evaluated positions. + + Returns: + number of rows written + """ + + if not movetext_lines: + return 0 + + headers = parse_headers(header_lines) + + site = headers.get("Site") + game_id = extract_game_id(site) + + if game_id is None: + print( + " [warning] Game has no valid Site header, skipping.", + file=sys.stderr + ) + return 0 + + movetext = " ".join( + line.strip() + for line in movetext_lines + if line.strip() + ) + + board = chess.Board() + rows_written = 0 + + for move_san, eval_val in tokenize_movetext(movetext): + + # No evaluation means this position has no target value. + # Do not put it into the regression dataset. + if eval_val is None: + continue + + clean_san = SUFFIX_ANNOTATION_RE.sub("", move_san) + + try: + board.push_san(clean_san) + + except ( + chess.IllegalMoveError, + chess.InvalidMoveError, + chess.AmbiguousMoveError, + ValueError, + ) as e: + + print( + f" [game {game_id}] stopping early: " + f"could not parse move '{move_san}' ({e})", + file=sys.stderr, + ) + + break + + turn = ( + "white" + if board.turn == chess.WHITE + else "black" + ) + + row = board_row(board) + + row += [ + eval_val, + turn, + ] + + if include_game_id: + row = [game_id] + row + + writer.writerow(row) + + rows_written += 1 + + return rows_written + + +def refine(input_path, output_path, include_game_id): + header_lines = [] + movetext_lines = [] + + state = "seek_header" + + games = 0 + total_rows = 0 + + header_row = SQUARE_COLUMNS + ["eval", "turn"] + + if include_game_id: + header_row = ["game_id"] + header_row + + with ( + open( + input_path, + "r", + encoding="utf-8", + errors="replace", + buffering=1024 * 1024, + ) as fin, + + open( + output_path, + "w", + encoding="utf-8", + newline="", + ) as fout + ): + + writer = csv.writer(fout) + writer.writerow(header_row) + + for raw_line in fin: + line = raw_line.rstrip("\n") + + if state == "seek_header": + + if line.startswith("["): + header_lines.append(line) + + elif line.strip() == "" and header_lines: + state = "seek_movetext" + + elif state == "seek_movetext": + + if line.strip() == "": + rows = process_game( + header_lines, + movetext_lines, + writer, + include_game_id, + ) + + total_rows += rows + games += 1 + + header_lines = [] + movetext_lines = [] + + state = "seek_header" + + else: + movetext_lines.append(line) + + # Handle final game if the file doesn't end with a blank line. + if header_lines and movetext_lines: + rows = process_game( + header_lines, + movetext_lines, + writer, + include_game_id, + ) + + total_rows += rows + games += 1 + + return games, total_rows + + +def main(): + parser = argparse.ArgumentParser( + description=( + "Convert refined PGN games into a " + "per-move board-state CSV." + ) + ) + + parser.add_argument( + "input", + help="Path to the refined PGN file", + ) + + parser.add_argument( + "output", + help="Path to write the CSV file", + ) + + parser.add_argument( + "--no-game-id", + action="store_true", + help="Omit the leading game_id column.", + ) + + args = parser.parse_args() + + games, rows = refine( + args.input, + args.output, + include_game_id=not args.no_game_id, + ) + + print( + f"Processed {games:,} games " + f"-> {rows:,} evaluated positions written to " + f"{args.output}" + ) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/cleaning/refine.py b/src/data_processing/refine.py similarity index 100% rename from cleaning/refine.py rename to src/data_processing/refine.py diff --git a/src/training/decision_tree/main.py b/src/training/decision_tree/main.py new file mode 100644 index 0000000..1d134eb --- /dev/null +++ b/src/training/decision_tree/main.py @@ -0,0 +1,71 @@ +import os + +from sklearn.ensemble import HistGradientBoostingRegressor, GradientBoostingRegressor, RandomForestRegressor +from sklearn.model_selection import train_test_split, GroupShuffleSplit +from sklearn.tree import DecisionTreeRegressor + +from src.training.decision_tree import preprocessor +from src.training.decision_tree.model import Model +from src.training.decision_tree.visualizer import Visualizer + + +def create_split(x, y, game_ids): + + splitter = GroupShuffleSplit( + n_splits=1, + test_size=0.10, + random_state=42 + ) + + train_indices, test_indices = next(splitter.split(x, y, groups=game_ids)) + + x_train = x[train_indices] + x_test = x[test_indices] + + y_train = y[train_indices] + y_test = y[test_indices] + + train_games = set(game_ids[train_indices]) + test_games = set(game_ids[test_indices]) + + print("Train positions:", len(x_train)) + print("Test positions:", len(x_test)) + print("Train games:", len(train_games)) + print("Test games:", len(test_games)) + print("Overlapping games:", len(train_games & test_games)) + + return x_train, x_test, y_train, y_test + +if __name__ == "__main__": + + model_path = "C:/Users/Laurent/Desktop/chess-engine/src/training/decision_tree/cache/model.joblib" + + if os.path.exists(model_path): + model = Model.load(model_path) + + else: + x, y, game_ids = preprocessor.run("C:/Users/Laurent/Desktop/chess-engine/dataset/v1/output.csv") + + split = create_split(x, y, game_ids) + + depth = 35 + leafs = int(depth * 3.5) + + model = Model( + tree=RandomForestRegressor( + n_estimators=25, + max_depth=depth, + max_features=0.35, + min_samples_leaf=leafs, + random_state=42 + ), + feature_names=preprocessor.get_feature_names(), + train_test_split=split + ) + + model.train() + model.evaluate() + #model.save_model(model_path) + + visualizer = Visualizer(model=model) + visualizer.show_feature_importance(15) diff --git a/src/training/decision_tree/model.py b/src/training/decision_tree/model.py new file mode 100644 index 0000000..504612a --- /dev/null +++ b/src/training/decision_tree/model.py @@ -0,0 +1,75 @@ +import os +import joblib +from sklearn.metrics import mean_absolute_error, mean_squared_error + + +class Model: + def __init__(self, tree=None, feature_names=None, train_test_split=None): + self._model = tree + self._feature_names = feature_names + self._x_train = self._x_test = None + self._y_train = self._y_test = None + + if train_test_split is not None: + self._x_train, self._x_test, \ + self._y_train, self._y_test = train_test_split + + @classmethod + def load(cls, path): + data = joblib.load(path) + + model = cls() + model._model = data["model"] + model._feature_names = data["feature_names"] + + print(f"Loaded model from {path}") + + return model + + def get_feature_names(self): + return self._feature_names + + def get_feature_importance(self): + importances = self._model.feature_importances_ + + return sorted( + zip(self._feature_names, importances), + key=lambda x: x[1], + reverse=True + ) + + def train(self): + print("Start training...") + self._model.fit(self._x_train, self._y_train) + print("Training complete.") + + def evaluate(self): + print("Evaluating model...") + + train_predictions = self._model.predict(self._x_train) + test_predictions = self._model.predict(self._x_test) + + train_mae = mean_absolute_error(self._y_train, train_predictions) + train_mse = mean_squared_error(self._y_train, train_predictions) + + test_mae = mean_absolute_error(self._y_test, test_predictions) + test_mse = mean_squared_error(self._y_test, test_predictions) + + print() + print("Training:") + print("MAE:", train_mae) + print("MSE:", train_mse) + + print() + print("Test:") + print("MAE:", test_mae) + print("MSE:", test_mse) + + + def save_model(self, path): + joblib.dump({ + "model": self._model, + "feature_names": self._feature_names + }, path) + + print(f"Saved model to {path}") \ No newline at end of file diff --git a/src/training/decision_tree/preprocessor.py b/src/training/decision_tree/preprocessor.py new file mode 100644 index 0000000..5b10681 --- /dev/null +++ b/src/training/decision_tree/preprocessor.py @@ -0,0 +1,1030 @@ +import os +import chess +import numpy as np +import pandas as pd + +PIECE_TO_PLANE = { + "P": 0, "N": 1, "B": 2, "R": 3, "Q": 4, "K": 5, + "p": 6, "n": 7, "b": 8, "r": 9, "q": 10, "k": 11 +} + +PIECE_VALUES = { + "P": 1, "N": 3, "B": 3, "R": 5, "Q": 9, + "p": 1, "n": 3, "b": 3, "r": 5, "q": 9 +} + +def parse_eval(value): + value = str(value).strip() + + if value.startswith("#"): + mate_distance = int(value[1:]) + return 10.0 if mate_distance > 0 else -10.0 + + return float(value) + + +def get_board(row): + board = chess.Board.empty() + + for square, piece_char in row.items(): + if square not in chess.SQUARE_NAMES: + continue + + if pd.isna(piece_char): + continue + + piece_char = str(piece_char).strip() + + if piece_char == "": + continue + + piece_type = chess.Piece.from_symbol(piece_char) + board.set_piece_at(chess.parse_square(square), piece_type) + + if row["turn"] == "black": + board.turn = chess.BLACK + else: + board.turn = chess.WHITE + + return board + + +def get_side_to_move_features(board): + return [1] if board.turn == chess.WHITE else [0] + +def get_material_features(board): + white = {piece: 0 for piece in "PNBRQ"} + black = {piece: 0 for piece in "pnbrq"} + + for piece in board.piece_map().values(): + symbol = piece.symbol() + + if symbol in white: + white[symbol] += 1 + elif symbol in black: + black[symbol] += 1 + + white_pawns = white["P"] + white_knights = white["N"] + white_bishops = white["B"] + white_rooks = white["R"] + white_queens = white["Q"] + + black_pawns = black["p"] + black_knights = black["n"] + black_bishops = black["b"] + black_rooks = black["r"] + black_queens = black["q"] + + white_value = ( + white_pawns + + white_knights * 3 + + white_bishops * 3 + + white_rooks * 5 + + white_queens * 9 + ) + + black_value = ( + black_pawns + + black_knights * 3 + + black_bishops * 3 + + black_rooks * 5 + + black_queens * 9 + ) + + return [ + white_pawns, + white_knights, + white_bishops, + white_rooks, + white_queens, + black_pawns, + black_knights, + black_bishops, + black_rooks, + black_queens, + white_value - black_value + ] + + +def get_max_mobility(board, square, piece): + piece_type = piece.piece_type + + if piece_type == chess.KNIGHT: + return len(chess.SquareSet(chess.BB_KNIGHT_ATTACKS[square])) + + if piece_type == chess.KING: + return len(chess.SquareSet(chess.BB_KING_ATTACKS[square])) + + rank = chess.square_rank(square) + file = chess.square_file(square) + + if piece_type == chess.ROOK: + return (7 - file) + file + (7 - rank) + rank + + if piece_type == chess.BISHOP: + return min(file, rank) + min(7 - file, rank) + min(file, 7 - rank) + min(7 - file, 7 - rank) + + if piece_type == chess.QUEEN: + rook_max = (7 - file) + file + (7 - rank) + rank + bishop_max = ( + min(file, rank) + + min(7 - file, rank) + + min(file, 7 - rank) + + min(7 - file, 7 - rank) + ) + return rook_max + bishop_max + + return 1 + + +def get_mobility_features(board): + mobility = { + "white_knight": 0, + "black_knight": 0, + "white_bishop": 0, + "black_bishop": 0, + "white_rook": 0, + "black_rook": 0, + "white_queen": 0, + "black_queen": 0, + "white_king": 0, + "black_king": 0 + } + + mobility_max = { + key: 0 + for key in mobility + } + + piece_names = { + chess.KNIGHT: "knight", + chess.BISHOP: "bishop", + chess.ROOK: "rook", + chess.QUEEN: "queen", + chess.KING: "king" + } + + for square, piece in board.piece_map().items(): + piece_name = piece_names.get(piece.piece_type) + + if piece_name is None: + continue + + color = "white" if piece.color == chess.WHITE else "black" + key = f"{color}_{piece_name}" + + actual = len(board.attacks(square)) + maximum = get_max_mobility(board, square, piece) + + mobility[key] += actual + mobility_max[key] += maximum + + ratios = {} + + for key in mobility: + if mobility_max[key] == 0: + ratios[key] = 0.0 + else: + ratios[key] = mobility[key] / mobility_max[key] + + return [ + mobility["white_knight"], + mobility["black_knight"], + mobility["white_knight"] - mobility["black_knight"], + ratios["white_knight"], + ratios["black_knight"], + ratios["white_knight"] - ratios["black_knight"], + + mobility["white_bishop"], + mobility["black_bishop"], + mobility["white_bishop"] - mobility["black_bishop"], + ratios["white_bishop"], + ratios["black_bishop"], + ratios["white_bishop"] - ratios["black_bishop"], + + mobility["white_rook"], + mobility["black_rook"], + mobility["white_rook"] - mobility["black_rook"], + ratios["white_rook"], + ratios["black_rook"], + ratios["white_rook"] - ratios["black_rook"], + + mobility["white_queen"], + mobility["black_queen"], + mobility["white_queen"] - mobility["black_queen"], + ratios["white_queen"], + ratios["black_queen"], + ratios["white_queen"] - ratios["black_queen"], + + mobility["white_king"], + mobility["black_king"], + mobility["white_king"] - mobility["black_king"], + ratios["white_king"], + ratios["black_king"], + ratios["white_king"] - ratios["black_king"], + + sum(mobility.values()) + ] + + +def count_doubled_pawns(board, color): + files = [0] * 8 + + for square in board.pieces(chess.PAWN, color): + files[chess.square_file(square)] += 1 + + return sum(max(0, count - 1) for count in files) + + +def count_isolated_pawns(board, color): + files = [0] * 8 + + for square in board.pieces(chess.PAWN, color): + files[chess.square_file(square)] += 1 + + count = 0 + + for file in range(8): + if files[file] == 0: + continue + + left = files[file - 1] if file > 0 else 0 + right = files[file + 1] if file < 7 else 0 + + if left == 0 and right == 0: + count += files[file] + + return count + + +def count_passed_pawns(board, color): + count = 0 + enemy = not color + + for square in board.pieces(chess.PAWN, color): + file = chess.square_file(square) + rank = chess.square_rank(square) + + passed = True + + for enemy_square in board.pieces(chess.PAWN, enemy): + enemy_file = chess.square_file(enemy_square) + enemy_rank = chess.square_rank(enemy_square) + + if abs(enemy_file - file) > 1: + continue + + if color == chess.WHITE and enemy_rank > rank: + passed = False + break + + if color == chess.BLACK and enemy_rank < rank: + passed = False + break + + if passed: + count += 1 + + return count + + +def get_pawn_structure_features(board): + white_doubled = count_doubled_pawns(board, chess.WHITE) + black_doubled = count_doubled_pawns(board, chess.BLACK) + + white_isolated = count_isolated_pawns(board, chess.WHITE) + black_isolated = count_isolated_pawns(board, chess.BLACK) + + white_passed = count_passed_pawns(board, chess.WHITE) + black_passed = count_passed_pawns(board, chess.BLACK) + + return [ + white_doubled, + black_doubled, + white_isolated, + black_isolated, + white_passed, + black_passed + ] + + +def count_king_attackers(board, color): + king = board.king(color) + + if king is None: + return 0 + + enemy = not color + count = 0 + + for attacker in board.attackers(enemy, king): + count += 1 + + return count + + +def count_king_escape_squares(board, color): + king = board.king(color) + + if king is None: + return 0 + + count = 0 + + for square in board.attacks(king): + piece = board.piece_at(square) + + if piece is not None and piece.color == color: + continue + + if not board.is_attacked_by(not color, square): + count += 1 + + return count + + +def get_pawn_shield(board, color): + king = board.king(color) + + if king is None: + return 0 + + king_file = chess.square_file(king) + king_rank = chess.square_rank(king) + direction = 1 if color == chess.WHITE else -1 + shield_rank = king_rank + direction + + if shield_rank < 0 or shield_rank > 7: + return 0 + + count = 0 + + for file in range(max(0, king_file - 1), min(7, king_file + 1) + 1): + square = chess.square(file, shield_rank) + + if board.piece_at(square) == chess.Piece(chess.PAWN, color): + count += 1 + + return count + + +def get_king_safety_features(board): + white_attackers = count_king_attackers(board, chess.WHITE) + black_attackers = count_king_attackers(board, chess.BLACK) + + white_escape = count_king_escape_squares(board, chess.WHITE) + black_escape = count_king_escape_squares(board, chess.BLACK) + + white_king = board.king(chess.WHITE) + black_king = board.king(chess.BLACK) + + white_castled = int(white_king in [chess.C1, chess.G1]) + black_castled = int(black_king in [chess.C8, chess.G8]) + + white_shield = get_pawn_shield(board, chess.WHITE) + black_shield = get_pawn_shield(board, chess.BLACK) + + return [ + white_attackers, + black_attackers, + white_escape, + black_escape, + white_castled, + black_castled, + white_shield, + black_shield + ] + + +def get_activity_features(board): + return [] + + +def get_center_control_features(board): + center = [chess.D4, chess.E4, chess.D5, chess.E5] + + white = sum(len(board.attackers(chess.WHITE, square)) for square in center) + black = sum(len(board.attackers(chess.BLACK, square)) for square in center) + + return [ + white, + black, + white - black + ] + + +def get_development_features(board): + white_developed = 0 + black_developed = 0 + + white_start = { + chess.B1, chess.G1, + chess.C1, chess.F1 + } + + black_start = { + chess.B8, chess.G8, + chess.C8, chess.F8 + } + + for square in board.pieces(chess.KNIGHT, chess.WHITE): + if square not in white_start: + white_developed += 1 + + for square in board.pieces(chess.BISHOP, chess.WHITE): + if square not in white_start: + white_developed += 1 + + for square in board.pieces(chess.KNIGHT, chess.BLACK): + if square not in black_start: + black_developed += 1 + + for square in board.pieces(chess.BISHOP, chess.BLACK): + if square not in black_start: + black_developed += 1 + + white_bishop_pair = int(len(board.pieces(chess.BISHOP, chess.WHITE)) >= 2) + black_bishop_pair = int(len(board.pieces(chess.BISHOP, chess.BLACK)) >= 2) + + return [ + white_developed, + black_developed, + white_developed - black_developed, + white_bishop_pair, + black_bishop_pair, + white_bishop_pair - black_bishop_pair + ] + + +def get_tactical_features(board): + white_attacked_pieces = 0 + black_attacked_pieces = 0 + white_hanging_pieces = 0 + black_hanging_pieces = 0 + + white_attacked_value = 0 + black_attacked_value = 0 + + for square, piece in board.piece_map().items(): + if piece.piece_type == chess.KING: + continue + + enemy = not piece.color + attackers = board.attackers(enemy, square) + defenders = board.attackers(piece.color, square) + + if attackers: + if piece.color == chess.WHITE: + white_attacked_pieces += 1 + white_attacked_value += PIECE_VALUES[piece.symbol()] + else: + black_attacked_pieces += 1 + black_attacked_value += PIECE_VALUES[piece.symbol()] + + if attackers and not defenders: + if piece.color == chess.WHITE: + white_hanging_pieces += 1 + else: + black_hanging_pieces += 1 + + return [ + white_attacked_pieces, + black_attacked_pieces, + white_attacked_pieces - black_attacked_pieces, + + white_hanging_pieces, + black_hanging_pieces, + white_hanging_pieces - black_hanging_pieces, + + white_attacked_value, + black_attacked_value, + white_attacked_value - black_attacked_value + ] + +def get_pin_features(board): + white_pinned = 0 + black_pinned = 0 + white_pinned_value = 0 + black_pinned_value = 0 + + for square, piece in board.piece_map().items(): + if piece.piece_type == chess.KING: + continue + + if board.is_pinned(piece.color, square): + value = PIECE_VALUES.get(piece.symbol(), 0) + + if piece.color == chess.WHITE: + white_pinned += 1 + white_pinned_value += value + else: + black_pinned += 1 + black_pinned_value += value + + return [ + white_pinned, + black_pinned, + white_pinned - black_pinned, + white_pinned_value, + black_pinned_value, + white_pinned_value - black_pinned_value + ] + + +def get_ray_directions(piece_type): + if piece_type == chess.ROOK: + return [(1, 0), (-1, 0), (0, 1), (0, -1)] + + if piece_type == chess.BISHOP: + return [(1, 1), (1, -1), (-1, 1), (-1, -1)] + + if piece_type == chess.QUEEN: + return [ + (1, 0), (-1, 0), (0, 1), (0, -1), + (1, 1), (1, -1), (-1, 1), (-1, -1) + ] + + return [] + + +def get_skewer_features(board): + white_skewer_value = 0 + black_skewer_value = 0 + + for square, piece in board.piece_map().items(): + if piece.piece_type not in [chess.ROOK, chess.BISHOP, chess.QUEEN]: + continue + + directions = get_ray_directions(piece.piece_type) + file = chess.square_file(square) + rank = chess.square_rank(square) + + for df, dr in directions: + f = file + df + r = rank + dr + + first = None + second = None + + while 0 <= f < 8 and 0 <= r < 8: + target_square = chess.square(f, r) + target_piece = board.piece_at(target_square) + + if target_piece is not None: + if first is None: + first = target_piece + else: + second = target_piece + break + + f += df + r += dr + + if first is None or second is None: + continue + + if first.color != piece.color or second.color != piece.color: + continue + + # The second piece must be more valuable than the first. + first_value = PIECE_VALUES.get(first.symbol(), 0) + second_value = PIECE_VALUES.get(second.symbol(), 0) + + if second_value <= first_value: + continue + + if piece.color == chess.WHITE: + white_skewer_value += second_value + else: + black_skewer_value += second_value + + return [ + white_skewer_value, + black_skewer_value, + white_skewer_value - black_skewer_value + ] + +def get_discovered_attack_features(board): + white_value = 0 + black_value = 0 + + for slider_square, slider in board.piece_map().items(): + if slider.piece_type not in [chess.ROOK, chess.BISHOP, chess.QUEEN]: + continue + + directions = get_ray_directions(slider.piece_type) + slider_file = chess.square_file(slider_square) + slider_rank = chess.square_rank(slider_square) + + for df, dr in directions: + f = slider_file + df + r = slider_rank + dr + + blocker = None + + while 0 <= f < 8 and 0 <= r < 8: + square = chess.square(f, r) + piece = board.piece_at(square) + + if piece is not None: + blocker = piece + blocker_square = square + break + + f += df + r += dr + + if blocker is None or blocker.color != slider.color: + continue + + if blocker.piece_type == chess.KING: + continue + + f += df + r += dr + + while 0 <= f < 8 and 0 <= r < 8: + square = chess.square(f, r) + target = board.piece_at(square) + + if target is not None: + if target.color != slider.color: + value = 9 if target.piece_type == chess.KING else PIECE_VALUES.get(target.symbol(), 0) + + if slider.color == chess.WHITE: + white_value += value + else: + black_value += value + + break + + f += df + r += dr + + return [ + white_value, + black_value, + white_value - black_value + ] + + +def get_overloaded_defender_features(board): + white_value = 0 + black_value = 0 + + for defender_square, defender in board.piece_map().items(): + if defender.piece_type == chess.KING: + continue + + targets = [] + + for target_square, target in board.piece_map().items(): + if target.color != defender.color: + continue + + if target_square == defender_square: + continue + + if target.piece_type == chess.KING: + continue + + if defender_square in board.attackers(defender.color, target_square): + targets.append(PIECE_VALUES.get(target.symbol(), 0)) + + if len(targets) < 2: + continue + + # Only count meaningful overloaded defenders. + targets.sort(reverse=True) + defended_value = sum(targets) + + if defended_value < 6: + continue + + if defender.color == chess.WHITE: + white_value += defended_value + else: + black_value += defended_value + + return [ + white_value, + black_value, + white_value - black_value + ] + + +def get_king_attack_value_features(board): + white_attack_value = 0 + black_attack_value = 0 + + white_king = board.king(chess.WHITE) + black_king = board.king(chess.BLACK) + + if white_king is not None: + for square in board.attackers(chess.BLACK, white_king): + piece = board.piece_at(square) + + if piece is not None: + black_attack_value += PIECE_VALUES.get(piece.symbol(), 0) + + if black_king is not None: + for square in board.attackers(chess.WHITE, black_king): + piece = board.piece_at(square) + + if piece is not None: + white_attack_value += PIECE_VALUES.get(piece.symbol(), 0) + + return [ + white_attack_value, + black_attack_value, + white_attack_value - black_attack_value + ] + + +def get_important_square_features(board): + squares = { + chess.D4, chess.E4, + chess.D5, chess.E5 + } + + for color in [chess.WHITE, chess.BLACK]: + king = board.king(color) + + if king is None: + continue + + king_file = chess.square_file(king) + king_rank = chess.square_rank(king) + + for df in [-1, 0, 1]: + for dr in [-1, 0, 1]: + file = king_file + df + rank = king_rank + dr + + if 0 <= file < 8 and 0 <= rank < 8: + squares.add(chess.square(file, rank)) + + white_attackers = 0 + black_attackers = 0 + white_value = 0 + black_value = 0 + + for square in squares: + for attacker in board.attackers(chess.WHITE, square): + piece = board.piece_at(attacker) + + if piece is not None: + white_attackers += 1 + white_value += PIECE_VALUES.get(piece.symbol(), 0) + + for attacker in board.attackers(chess.BLACK, square): + piece = board.piece_at(attacker) + + if piece is not None: + black_attackers += 1 + black_value += PIECE_VALUES.get(piece.symbol(), 0) + + return [ + white_attackers, + black_attackers, + white_attackers - black_attackers, + white_value, + black_value, + white_value - black_value + ] + + +def get_immediate_capture_features(board): + white_capture_count = 0 + black_capture_count = 0 + white_capture_value = 0 + black_capture_value = 0 + + for color in [chess.WHITE, chess.BLACK]: + board.turn = color + + for move in board.legal_moves: + if not board.is_capture(move): + continue + + captured_piece = board.piece_at(move.to_square) + + # En-passant cannot be reconstructed from the current dataset. + if captured_piece is None: + continue + + value = PIECE_VALUES.get(captured_piece.symbol(), 0) + + if color == chess.WHITE: + white_capture_count += 1 + white_capture_value += value + else: + black_capture_count += 1 + black_capture_value += value + + board.turn = chess.WHITE + + return [ + white_capture_count, + black_capture_count, + white_capture_count - black_capture_count, + white_capture_value, + black_capture_value, + white_capture_value - black_capture_value + ] + + +def get_file_features(board): + white_pawns = [0] * 8 + black_pawns = [0] * 8 + + for square in board.pieces(chess.PAWN, chess.WHITE): + white_pawns[chess.square_file(square)] += 1 + + for square in board.pieces(chess.PAWN, chess.BLACK): + black_pawns[chess.square_file(square)] += 1 + + white_open_rooks = 0 + black_open_rooks = 0 + white_semi_open_rooks = 0 + black_semi_open_rooks = 0 + + for square in board.pieces(chess.ROOK, chess.WHITE): + file = chess.square_file(square) + + if white_pawns[file] == 0 and black_pawns[file] == 0: + white_open_rooks += 1 + elif white_pawns[file] == 0: + white_semi_open_rooks += 1 + + for square in board.pieces(chess.ROOK, chess.BLACK): + file = chess.square_file(square) + + if white_pawns[file] == 0 and black_pawns[file] == 0: + black_open_rooks += 1 + elif black_pawns[file] == 0: + black_semi_open_rooks += 1 + + return [ + white_open_rooks, + black_open_rooks, + white_open_rooks - black_open_rooks, + white_semi_open_rooks, + black_semi_open_rooks, + white_semi_open_rooks - black_semi_open_rooks + ] + + +def get_feature_names(): + return [ + #General + "side_to_move", + + # Material + "white_pawns", "white_knights", "white_bishops", "white_rooks", "white_queens", + "black_pawns", "black_knights", "black_bishops", "black_rooks", "black_queens", + "material_balance", + + # Mobility + "white_knight_mobility", "black_knight_mobility", "knight_mobility_balance", + "white_knight_mobility_ratio", "black_knight_mobility_ratio", "knight_mobility_ratio_balance", + "white_bishop_mobility", "black_bishop_mobility", "bishop_mobility_balance", + "white_bishop_mobility_ratio", "black_bishop_mobility_ratio", "bishop_mobility_ratio_balance", + "white_rook_mobility", "black_rook_mobility", "rook_mobility_balance", + "white_rook_mobility_ratio", "black_rook_mobility_ratio", "rook_mobility_ratio_balance", + "white_queen_mobility", "black_queen_mobility", "queen_mobility_balance", + "white_queen_mobility_ratio", "black_queen_mobility_ratio", "queen_mobility_ratio_balance", + "white_king_mobility", "black_king_mobility", "king_mobility_balance", + "white_king_mobility_ratio", "black_king_mobility_ratio", "king_mobility_ratio_balance", + "total_mobility", + + # Pawn structure + "white_doubled_pawns", "black_doubled_pawns", + "white_isolated_pawns", "black_isolated_pawns", + "white_passed_pawns", "black_passed_pawns", + + # King safety + "white_king_attackers", "black_king_attackers", + "white_king_escape_squares", "black_king_escape_squares", + "white_castled", "black_castled", + "white_pawn_shield", "black_pawn_shield", + + # Development + "white_developed_minors", "black_developed_minors", "development_balance", + "white_bishop_pair", "black_bishop_pair", "bishop_pair_balance", + + # Center control + "white_center_control", "black_center_control", "center_control_balance", + + # Tactical pressure + "white_attacked_pieces", "black_attacked_pieces", "attacked_pieces_balance", + "white_hanging_pieces", "black_hanging_pieces", "hanging_pieces_balance", + "white_attacked_value", "black_attacked_value", "attacked_value_balance", + + # Rooks and files + "white_rooks_open_files", "black_rooks_open_files", "open_rooks_balance", + "white_rooks_semi_open_files", "black_rooks_semi_open_files", "semi_open_rooks_balance", + + # Tactical patterns + "white_pinned_pieces", "black_pinned_pieces", + "pinned_pieces_balance", + "white_pinned_value", "black_pinned_value", "pinned_value_balance", + + "white_skewer_value", "black_skewer_value", "skewer_value_balance", + + "white_discovered_attack_value", "black_discovered_attack_value", + "discovered_attack_value_balance", + + "white_overloaded_defender_value", "black_overloaded_defender_value", + "overloaded_defender_value_balance", + + "white_king_attack_value", "black_king_attack_value", + "king_attack_value_balance", + + "white_important_square_attackers", "black_important_square_attackers", + "important_square_attackers_balance", + "white_important_square_attack_value", "black_important_square_attack_value", + "important_square_attack_value_balance", + + "white_immediate_captures", "black_immediate_captures", + "immediate_captures_balance", + "white_immediate_capture_value", "black_immediate_capture_value", + "immediate_capture_value_balance" + ] + + +def board_to_features(row): + board = get_board(row) + + features = [] + features.extend(get_side_to_move_features(board)) + features.extend(get_material_features(board)) + features.extend(get_mobility_features(board)) + features.extend(get_pawn_structure_features(board)) + features.extend(get_king_safety_features(board)) + features.extend(get_development_features(board)) + features.extend(get_center_control_features(board)) + features.extend(get_tactical_features(board)) + features.extend(get_file_features(board)) + + features.extend(get_pin_features(board)) + features.extend(get_skewer_features(board)) + features.extend(get_discovered_attack_features(board)) + features.extend(get_overloaded_defender_features(board)) + features.extend(get_king_attack_value_features(board)) + features.extend(get_important_square_features(board)) + features.extend(get_immediate_capture_features(board)) + + return np.array(features, dtype=np.float32) + + +def run(path): + print("Preprocessing data...") + + cache_dir = os.path.join(os.path.dirname(path), "cache") + os.makedirs(cache_dir, exist_ok=True) + + x_path = os.path.join(cache_dir, "x.npy") + y_path = os.path.join(cache_dir, "y.npy") + game_ids_path = os.path.join(cache_dir, "game_ids.npy") + + if os.path.exists(x_path) and os.path.exists(y_path) and os.path.exists(game_ids_path): + print("Loading preprocessed data from cache...") + return ( + np.load(x_path), + np.load(y_path), + np.load(game_ids_path, allow_pickle=True) + ) + + df = pd.read_csv(path) + + x = [] + y = [] + game_ids = [] + + max_index = len(df) - 1 + + for current_index, (_, row) in enumerate(df.iterrows()): + if current_index % 10000 == 0: + print(f"{current_index}/{max_index} ({current_index / max_index:.2%})") + + x.append(board_to_features(row)) + y.append(parse_eval(row["eval"])) + game_ids.append(row["game_id"]) + + x = np.asarray(x, dtype=np.float32) + y = np.asarray(y, dtype=np.float32) + game_ids = np.asarray(game_ids) + + print("Saving preprocessed data to cache...") + + np.save(x_path, x) + np.save(y_path, y) + np.save(game_ids_path, game_ids) + + print("NaN evaluations:", np.isnan(y).sum()) + print("X shape:", x.shape) + print("y shape:", y.shape) + + return x, y, game_ids \ No newline at end of file diff --git a/src/training/decision_tree/visualizer.py b/src/training/decision_tree/visualizer.py new file mode 100644 index 0000000..cfccac7 --- /dev/null +++ b/src/training/decision_tree/visualizer.py @@ -0,0 +1,50 @@ +import chess +import numpy as np +from matplotlib import pyplot as plt + + +class Visualizer: + def __init__(self, model): + self._model = model + + def show_feature_importance(self, top_n=15): + importances = self._model.get_feature_importance() + print("Feature importance") + print("-------------------------") + + for feature_name, importance in importances[:top_n]: + print(f"{feature_name:<35} {importance:.6f}") + + def show_piece_heatmaps(self): + importances = dict(self._model.get_feature_importance()) + + pieces = [ + "P", "N", "B", "R", "Q", "K", + "p", "n", "b", "r", "q", "k" + ] + + fig, axes = plt.subplots(3, 4, figsize=(16, 12)) + + for ax, piece in zip(axes.flat, pieces): + board = np.zeros((8, 8)) + + for row, rank in enumerate(range(7, -1, -1)): + for col, file in enumerate(range(8)): + square = chess.square_name(chess.square(file, rank)) + feature_name = f"{piece}_{square}" + board[row, col] = importances.get(feature_name, 0) + + image = ax.imshow(board) + + ax.set_xticks(range(8)) + ax.set_xticklabels("abcdefgh") + ax.set_yticks(range(8)) + ax.set_yticklabels("87654321") + ax.set_title(piece) + + fig.colorbar(image, ax=ax, fraction=0.046, pad=0.04) + + fig.suptitle("Piece-square feature importance", fontsize=16) + plt.tight_layout() + plt.show() + input() \ No newline at end of file