Simple chess AI

requirements.txt

@@ -1,5 +1,7 @@
 python-chess
-
+scikit-learn
 torch
 pandas
-pandas-stubs
+pandas-stubs
+matplotlib
+safetensors

src/inference/simpleChessAI.py

@@ -0,0 +1,85 @@
+import torch
+import chess
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+class LightweightFlyInference:
+    def __init__(self, model_name="lazy-guy12/chess-llama"):
+        self.model_name = model_name
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        
+        print(f"Auto-downloading and loading '{self.model_name}' from Hugging Face...")
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+        self.model = AutoModelForCausalLM.from_pretrained(self.model_name).to(self.device)
+        self.model.eval()
+        
+        # Pre-cache vocabulary token IDs to valid python-chess Move objects
+        self._build_vocab_move_cache()
+        print("Model successfully loaded and ready for inference.")
+
+    def _build_vocab_move_cache(self):
+        """Maps chess-llama's custom vocabulary tokens to valid python-chess Move objects."""
+        self.vocab_map = {}
+        for idx in range(len(self.tokenizer)):
+            token_str = self.tokenizer.decode([idx]).strip()
+            token_str = token_str.split()[0] if token_str.split() else ""
+            try:
+                if len(token_str) >= 4:
+                    move = chess.Move.from_uci(token_str)
+                    self.vocab_map[idx] = move
+            except ValueError:
+                continue
+
+    def forward(self, board: chess.Board) -> chess.Move:
+        """
+        Evaluates board state using model logits and selects the highest-probability 
+        legal move from the model's vocabulary distribution.
+        """
+        move_uci_list = [move.uci() for move in board.move_stack]
+        
+        # Handle FEN initialization check if move stack is empty
+        if not move_uci_list and board.fen().startswith("rnbqkbnr/pppppppp/8/8/4P3/8/PPPP1PPP/RNBQKBNR"):
+            move_uci_list = ["e2e4"]
+
+        prompt = " ".join(move_uci_list)
+        if prompt:
+            prompt += " "
+
+        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
+
+        with torch.no_grad():
+            # Get next-token logits directly from the model forward pass
+            outputs = self.model(**inputs)
+            next_token_logits = outputs.logits[0, -1, :] # Shape: (vocab_size,)
+
+        # Sort vocabulary indices by model confidence score (descending order)
+        sorted_logits, sorted_indices = torch.sort(next_token_logits, descending=True)
+
+        # Iterate through the model's predictions from most confident to least confident,
+        # and pick the first one that is completely legal for the current board state.
+        for idx in sorted_indices.tolist():
+            if idx in self.vocab_map:
+                move = self.vocab_map[idx]
+                if move in board.legal_moves:
+                    return move
+
+        # Ultimate fallback to the first legal move if no vocabulary match is found
+        print("Warning: No valid move found in model vocabulary. Falling back to first legal move.")
+        legal_moves = list(board.legal_moves)
+        if legal_moves:
+            return legal_moves[0]
+        
+        return None
+
+if __name__ == "__main__":
+    inference = LightweightFlyInference()
+
+    board = chess.Board("rnbqkbnr/pppppppp/8/8/4P3/8/PPPP1PPP/RNBQKBNR b KQkq - 0 1")
+    best_move = inference.forward(board)
+    
+    if best_move:
+        print(f"Recommended Move (UCI): {best_move.uci()}")
+        print(f"Recommended Move (SAN): {board.san(best_move)}")
+        board.push(best_move)
+        print("\nUpdated Chess Board:\n", board)
+    else:
+        print("No legal moves available.")

src/training/decision_tree/main.py

@@ -1,60 +1,23 @@
 import os
-import time
+import matplotlib.pyplot as plt
-from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+import numpy as np
-
+import pandas as pd
-from sklearn.ensemble import RandomForestRegressor
+import seaborn as sns
+from sklearn.ensemble import HistGradientBoostingRegressor
 from sklearn.model_selection import GroupShuffleSplit
 
 from src.training.decision_tree import preprocessor
 from src.training.decision_tree.model import Model
 from src.training.decision_tree.visualizer import Visualizer
+from src.training.simulated_annealing.simulated_annealing import SimulatedAnnealing
 
-import matplotlib.pyplot as plt
+# Path setup for saving figures
-import numpy as np
+SAVE_DIR = "./doc/hyperparametersOptimisation/simulatedAnnealing/"
-
+SAVE_PATH = os.path.join(SAVE_DIR, "pairwise_exploration.png")
-
+os.makedirs(SAVE_DIR, exist_ok=True)
-def get_axis_limits(values):
-    minimum = min(values)
-    maximum = max(values)
-
-    if minimum == maximum:
-        padding = abs(minimum) * 0.05 or 0.5
-    else:
-        padding = (maximum - minimum) * 0.05
-
-    return minimum - padding, maximum + padding
-
-
-def train_model(parameter_1, parameter_2, parameter_1_name, parameter_2_name,
-                base_model_parameters, split):
-    start_time = time.perf_counter()
-    model_parameters = {
-        **base_model_parameters,
-        parameter_1_name: parameter_1,
-        parameter_2_name: parameter_2
-    }
-
-    model = Model(
-        tree=RandomForestRegressor(**model_parameters),
-        feature_names=preprocessor.get_feature_names(),
-        train_test_split=split
-    )
-
-    model.train()
-    train_mae, train_mse, test_mae, test_mse = model.evaluate()
-    model.release_training_data()
-
-    return {
-        "model": model,
-        "parameter_1": parameter_1,
-        "parameter_2": parameter_2,
-        "train_mae": train_mae,
-        "train_mse": train_mse,
-        "test_mae": test_mae,
-        "test_mse": test_mse,
-        "duration": time.perf_counter() - start_time
-    }
 
+# Shared history state
+evaluations_history = []
 
 def create_split(x, y, game_ids):
 
@@ -83,149 +46,124 @@ def create_split(x, y, game_ids):
 
     return x_train, x_test, y_train, y_test
 
-if __name__ == "__main__":
+# Graph for hyperparameter visualisation
+def update_live_pairwise_plot(history, save_path):
+    """Refreshes the live Seaborn pairplot and exports the updated figure to disk."""
+    df = pd.DataFrame(
+        [
+            {
+                "max_depth": e["params"]["max_depth"],
+                "min_samples_leaf": e["params"]["min_samples_leaf"],
+                # "n_estimators": e["params"]["n_estimators"],
+                "max_features": e["params"]["max_features"],
+                "test_mse": e["test_mse"],
+            }
+            for e in history
+        ]
+    )
 
+    # Close previous figures to avoid accumulating memory and orphaned windows
+    plt.close("all")
+
+    # Render pairwise exploration matrix
+    grid = sns.pairplot(
+        df,
+        vars=[
+            "max_depth",
+            "min_samples_leaf",
+            # "n_estimators",
+            "max_features"],
+        hue="test_mse",
+        palette="crest",
+        diag_kind="kde",
+    )
+    grid.figure.suptitle(f"Live SA Search (Iteration {len(history)})", y=1.02)
+
+    # Save to disk
+    grid.savefig(save_path, bbox_inches="tight")
+
+    # Force Matplotlib canvas redraw and flush event loop
+    grid.figure.canvas.draw()
+    grid.figure.canvas.flush_events()
+    plt.pause(0.05)
+
+# Callback function for training the model with given hyperparameters
+def train_model(parameters):
+    # Enable interactive figure mode on first run
+
+    model = Model(
+        tree=HistGradientBoostingRegressor(**parameters),
+        feature_names=preprocessor.get_feature_names(),
+        train_test_split=split,
+    )
+
+    model.train()
+    train_mae, train_mse, test_mae, test_mse = model.evaluate()
+    model.release_training_data()
+
+    # Track current iteration state
+    evaluations_history.append(
+        {
+            "params": parameters,
+            "train_mse": train_mse,
+            "test_mse": test_mse,
+        }
+    )
+
+    # Trigger live update & file saving
+    update_live_pairwise_plot(evaluations_history, SAVE_PATH)
+
+    return test_mse
+
+
+if __name__ == "__main__":
     model_path = "./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("./dataset/output.csv")
-
         split = create_split(x, y, game_ids)
         del x, y, game_ids
-
+        
-        parameter_1 = {
-            "name": "min_samples_leaf",
-            "label": "Leafs",
-            "values": list(range(16, 19))
-        }
-        parameter_2 = {
-            "name": "max_features",
-            "label": "Max features",
-            "values": [
-                round(value, 2)
-                for value in np.arange(0.45, 1, 0.05)
-            ]
-        }
-        base_model_parameters = {
-            "max_depth": 39,
-            "min_samples_leaf": 17,
-            "n_estimators": 26,
-            "max_features": 0.45,
-            "random_state": 42,
-            "n_jobs": 1
-        }
-
-        parameter_sets = [
-            (parameter_1_value, parameter_2_value)
-            for parameter_1_value in parameter_1["values"]
-            for parameter_2_value in parameter_2["values"]
-        ]
-        completed_parameter_1 = []
-        completed_parameter_2 = []
-        completed_maes = []
-
-        # Plot graph
-        fig, ax = plt.subplots()
-        scatter = ax.scatter([], [], c=[], cmap="RdYlGn_r", vmin=1.3, vmax=1.6)
-        colorbar = fig.colorbar(scatter, ax=ax)
-        colorbar.set_label("Test MAE")
-        ax.set_xlabel(parameter_1["label"])
-        ax.set_ylabel(parameter_2["label"])
-        ax.set_title(
-            "Random forest hyperparameter search "
-            f"({parameter_1['label']}, {parameter_2['label']})"
-        )
-        ax.set_xlim(get_axis_limits(parameter_1["values"]))
-        ax.set_ylim(get_axis_limits(parameter_2["values"]))
         plt.ion()
-        plt.show(block=False)
 
-        cpu_count = os.cpu_count() or 1
+        base_model_parameters = {
-        worker_count = max(1, min(cpu_count - 2, 8))
+            "max_depth": {"min": 1, "max": 100},
-        print(
+            "min_samples_leaf": {"min": 1, "max": 30},
-            f"Training {len(parameter_sets)} models with "
+            # "n_estimators": {"min": 10, "max": 100},
-            f"{worker_count} concurrent models..."
+            "max_features": {"min": 0.01, "max": 1.0},
+            "random_state": 42,
+            # "n_jobs": 15,
+        }
+
+        optimization_algorithm = SimulatedAnnealing(
+            parameter_space=base_model_parameters,
+            initial_temperature=1,
+            cooling_rate=0.90,
+            max_iterations=100,
         )
+        optimization_algorithm.set_training_callback(train_model)
 
-        with ThreadPoolExecutor(max_workers=worker_count) as executor:
+        try:
-            parameter_iterator = iter(parameter_sets)
+            best_solution, best_score = optimization_algorithm.optimize()
-            futures = {}
+        except KeyboardInterrupt:
+            print("\n[Optimization Interrupted] Early exit triggered by user.")
+            
+            
+        print("\n[Optimization Complete] Training model with best hyperparameters :")
+        model = Model(
+            tree=HistGradientBoostingRegressor(**best_solution),
+            feature_names=preprocessor.get_feature_names(),
+            train_test_split=split,
+        )
+    
+        model.train()
+        train_mae, train_mse, test_mae, test_mse = model.evaluate()
+        model.release_training_data()
 
-            def submit_next():
-                try:
-                    parameter_1_value, parameter_2_value = next(parameter_iterator)
-                except StopIteration:
-                    return
-
-                future = executor.submit(
-                    train_model,
-                    parameter_1_value,
-                    parameter_2_value,
-                    parameter_1["name"],
-                    parameter_2["name"],
-                    base_model_parameters,
-                    split
-                )
-                futures[future] = (parameter_1_value, parameter_2_value)
-
-            for _ in range(worker_count):
-                submit_next()
-
-            best_result = None
-            while futures:
-                completed_futures, _ = wait(
-                    futures,
-                    return_when=FIRST_COMPLETED
-                )
-
-                for future in completed_futures:
-                    futures.pop(future)
-                    result = future.result()
-                    completed_parameter_1.append(result["parameter_1"])
-                    completed_parameter_2.append(result["parameter_2"])
-                    completed_maes.append(result["test_mae"])
-
-                    scatter.set_offsets(np.column_stack((
-                        completed_parameter_1,
-                        completed_parameter_2
-                    )))
-                    scatter.set_array(np.asarray(completed_maes))
-                    fig.canvas.draw_idle()
-                    fig.canvas.flush_events()
-                    plt.pause(0.001)
-
-                    if (best_result is None or
-                            result["test_mae"] < best_result["test_mae"]):
-                        old_best_result = best_result
-                        best_result = result
-                        if old_best_result is not None:
-                            del old_best_result["model"]
-                    else:
-                        del result["model"]
-
-                    print(
-                        f"Completed {parameter_1['name']}={result['parameter_1']}, "
-                        f"{parameter_2['name']}={result['parameter_2']}. "
-                        f"Best so far: {parameter_1['name']}="
-                        f"{best_result['parameter_1']}, "
-                        f"{parameter_2['name']}={best_result['parameter_2']}, "
-                        f"MAE={best_result['test_mae']:.6f}, "
-                        f"duration={result['duration']:.1f}s"
-                    )
-
-                    submit_next()
-
-        model = best_result["model"]
         plt.ioff()
-        graph_path = f"doc/hyperparametersOptimisation/{parameter_1['name']}_{parameter_2['name']}.png"
+        plt.show()
-        os.makedirs(os.path.dirname(graph_path), exist_ok=True)
-        fig.savefig(graph_path, dpi=200, bbox_inches="tight")
-        print(f"Saved hyperparameter search graph to {graph_path}")
-        print("Close the graph window to finish.")
-        plt.show(block=True)
 
     visualizer = Visualizer(model=model)
-    visualizer.show_feature_importance(15)
+    visualizer.show_feature_importance(15)

src/training/simulated_annealing/simulated_annealing.py

@@ -0,0 +1,57 @@
+import random
+import math
+
+class SimulatedAnnealing:
+    def __init__(self, parameter_space, initial_temperature=1000, cooling_rate=0.95, max_iterations=50):
+        self.parameter_space = parameter_space
+        self.temperature = initial_temperature
+        self.cooling_rate = cooling_rate
+        self.max_iterations = max_iterations
+        
+    def random_solution(self):
+        solution = {}
+        for param, bounds in self.parameter_space.items():
+            # If bounds is a dict with min/max, generate bounded random values
+            if isinstance(bounds, dict) and "min" in bounds and "max" in bounds:
+                val = random.uniform(bounds["min"], bounds["max"])
+                
+                # Cast to integer if min/max are integers (e.g., max_depth, n_estimators)
+                if isinstance(bounds["min"], int) and isinstance(bounds["max"], int):
+                    val = int(round(val))
+                    
+                solution[param] = val
+            else:
+                # Pass fixed parameters straight through (e.g., random_state=42, n_jobs=1)
+                solution[param] = bounds
+                
+        return solution
+        
+    def set_training_callback(self, callback):
+        self.training_callback = callback
+        
+    def optimize(self):
+        current_solution = self.random_solution()
+        current_score = self.training_callback(current_solution)
+        
+        best_solution = current_solution
+        best_score = current_score
+        
+        for iteration in range(self.max_iterations):
+            new_solution = self.random_solution()
+            new_score = self.training_callback(new_solution)
+            
+            if new_score < best_score:
+                best_solution = new_solution
+                best_score = new_score
+            
+            acceptance_probability = math.exp((current_score - new_score) / self.temperature)
+            if new_score < current_score or random.random() < acceptance_probability:
+                current_solution = new_solution
+                current_score = new_score
+            
+            self.temperature *= self.cooling_rate
+            
+        print("Best solution:", best_solution)
+        print("Best score:", best_score)
+        
+        return best_solution, best_score