application_vnd.jupyter.stderr_output_2.png text_plain_output_1.png
from IPython.display import clear_output

from tensorflow.keras import layers, models, optimizers, losses

import chess

import numpy as np

import chess
import numpy as np
import tensorflow as tf
from tensorflow.keras import layers, models, optimizers, losses
from tensorflow.keras.layers import LeakyReLU
import concurrent.futures
from IPython.display import clear_output
import time
def create_chess_model():
    inputs = layers.Input(shape=(12, 8, 8))
    x = layers.Conv2D(32, (3, 3), padding='same', activation='relu')(inputs)
    x = layers.Conv2D(64, (3, 3), padding='same', activation='relu')(x)
    x = layers.Conv2D(8, (3, 3), padding='same', activation='relu')(x)
    x = layers.Attention(use_scale=True)([x, inputs])
    x = layers.Flatten()(x)
    x = layers.Dense(256, activation='relu')(x)
    x = layers.Dense(256, activation='relu')(x)
    outputs = layers.Dense(1)(x)
    model = models.Model(inputs=inputs, outputs=outputs)
    model.compile(optimizer=optimizers.Adam(learning_rate=0.001), loss=losses.MeanSquaredError())
    return model
def board_to_tensor(board):
    piece_map = board.piece_map()
    tensor = np.zeros((12, 8, 8), dtype=np.float32)
    for pos, piece in piece_map.items():
        x, y = (pos % 8, pos // 8)
        piece_index = piece.piece_type - 1 + (6 if piece.color == chess.BLACK else 0)
        tensor[piece_index, x, y] = 1
    return tensor[np.newaxis, :, :, :]
def evaluate(board, model):
    tensor = board_to_tensor(board)
    if model == chess_model_black:
        evaluations = -model(tensor).numpy().item()
    if model == chess_model_white:
        evaluations = model(tensor).numpy().item()
    return evaluations
def select_move(board, model, temperature=0.2):
    legal_moves = list(board.legal_moves)
    scores = []
    for move in legal_moves:
        board.push(move)
        scores.append(evaluate(board, model))
        board.pop()
    probs = np.exp(np.array(scores) / temperature)
    probs /= probs.sum()
    move_index = np.random.choice(len(legal_moves), p=probs)
    return legal_moves[move_index]
def play_single_game(model_white, model_black, game_id):
    board = chess.Board()
    game_moves = []
    game_values = []
    while not board.is_game_over():
        model = model_white if board.turn == chess.WHITE else model_black
        move = select_move(board, model)
        value = evaluate(board, model)
        game_moves.append(move)
        game_values.append(value)
        board.push(move)
        yield (game_id, board)
    result = board.result()
    target_values = np.zeros(len(game_values))
    if result == '1-0':
        target_values[-1] = 100
    elif result == '0-1':
        target_values[-1] = -100
    elif result == '1/2-1/2':
        target_values[-1] = 0
    for i in range(len(game_values) - 2, -1, -2):
        target_values[i] = -target_values[i + 1]
    train_data = []
    temp_board = chess.Board()
    for move in game_moves:
        temp_board.push(move)
        train_data.append(board_to_tensor(temp_board))
    train_data = np.vstack(train_data)
    yield (game_id, train_data, target_values)
def train_self_play(model_white, model_black, num_games=2, num_epochs=10000, num_workers=2):
    scores = {'white_wins': 0, 'black_wins': 0, 'draws': 0}
    for epoch in range(num_epochs):
        executor = concurrent.futures.ThreadPoolExecutor(max_workers=num_workers)
        game_progress = [play_single_game(model_white, model_black, game_id) for game_id in range(num_games)]
        finished_games = 0
        while finished_games < num_games:
            clear_output(wait=True)
            total_games = scores['white_wins'] + scores['black_wins'] + scores['draws']
            white_win_rate = 0
            black_win_rate = 0
            if scores['white_wins'] > 0:
                white_win_rate = scores['white_wins'] / total_games * 100
            if scores['black_wins'] > 0:
                black_win_rate = scores['black_wins'] / total_games * 100
            print(f'Epoch {epoch + 1}/{num_epochs}')
            print(f'Finished games: {finished_games}/{num_games}')
            print('Scores after epoch {}: {}'.format(epoch + 1, scores))
            print(f'White win rate: {white_win_rate:.2f}%')
            print(f'Black win rate: {black_win_rate:.2f}%')
            print()
            displayed_game = False
            for game_id, game_gen in enumerate(game_progress):
                if game_gen is None:
                    continue
                try:
                    game_state = next(game_gen)
                    if len(game_state) == 3:
                        game_id, train_data, target_values = game_state
                        model_white.fit(train_data[::2], target_values[::2], batch_size=len(target_values[::2]), verbose=0)
                        model_black.fit(train_data[1::2], target_values[1::2], batch_size=len(target_values[1::2]), verbose=0)
                        last_target_value = target_values[-1]
                        if last_target_value == 1:
                            scores['white_wins'] += 1
                        elif last_target_value == -1:
                            scores['black_wins'] += 1
                        else:
                            scores['draws'] += 1
                        game_progress[game_id] = None
                        finished_games += 1
                    elif game_id < 3 and (not displayed_game):
                        game_id, board = game_state
                        print(f'Game {game_id}:')
                        print(board)
                        print()
                        displayed_game = True
                except StopIteration:
                    game_progress[game_id] = None
        executor.shutdown(wait=True)
if __name__ == '__main__':
    chess_model_white = create_chess_model()
    chess_model_black = create_chess_model()
    train_self_play(chess_model_white, chess_model_black)