Create engine/search.py

engine/search.py

@@ -0,0 +1,294 @@
+"""
+Nexus-Core Search Engine
+Efficient alpha-beta with essential optimizations
+- Basic transposition table
+- Simple move ordering (MVV-LVA)
+- Quiescence search
+"""
+
+import onnxruntime as ort
+import numpy as np
+import chess
+import time
+import logging
+from pathlib import Path
+from typing import Optional, Dict, Tuple, List
+
+logger = logging.getLogger(__name__)
+
+
+class NexusCoreEngine:
+    """
+    Lightweight chess engine for Nexus-Core
+    Optimized for speed over strength
+    """
+    
+    PIECE_VALUES = {
+        chess.PAWN: 100,
+        chess.KNIGHT: 320,
+        chess.BISHOP: 330,
+        chess.ROOK: 500,
+        chess.QUEEN: 900,
+        chess.KING: 0
+    }
+    
+    def __init__(self, model_path: str, num_threads: int = 2):
+        """Initialize engine"""
+        
+        self.model_path = Path(model_path)
+        if not self.model_path.exists():
+            raise FileNotFoundError(f"Model not found: {model_path}")
+        
+        # Load ONNX model
+        sess_options = ort.SessionOptions()
+        sess_options.intra_op_num_threads = num_threads
+        sess_options.inter_op_num_threads = num_threads
+        sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+        
+        logger.info(f"Loading Nexus-Core from {model_path}...")
+        self.session = ort.InferenceSession(
+            str(self.model_path),
+            sess_options=sess_options,
+            providers=['CPUExecutionProvider']
+        )
+        
+        self.input_name = self.session.get_inputs()[0].name
+        self.output_name = self.session.get_outputs()[0].name
+        
+        # Simple transposition table (dict-based, 100K entries)
+        self.tt_cache = {}
+        self.max_tt_size = 100000
+        
+        # Statistics
+        self.nodes_evaluated = 0
+        
+        logger.info("✅ Nexus-Core engine ready")
+    
+    def fen_to_tensor(self, fen: str) -> np.ndarray:
+        """Convert FEN to 12-channel tensor"""
+        board = chess.Board(fen)
+        tensor = np.zeros((1, 12, 8, 8), dtype=np.float32)
+        
+        piece_to_channel = {
+            chess.PAWN: 0, chess.KNIGHT: 1, chess.BISHOP: 2,
+            chess.ROOK: 3, chess.QUEEN: 4, chess.KING: 5
+        }
+        
+        for square, piece in board.piece_map().items():
+            rank, file = divmod(square, 8)
+            channel = piece_to_channel[piece.piece_type]
+            if piece.color == chess.BLACK:
+                channel += 6
+            tensor[0, channel, rank, file] = 1.0
+        
+        return tensor
+    
+    def evaluate(self, board: chess.Board) -> float:
+        """Neural network evaluation"""
+        self.nodes_evaluated += 1
+        
+        # Check cache (simple FEN-based)
+        fen_key = board.fen().split(' ')[0]
+        if fen_key in self.tt_cache:
+            return self.tt_cache[fen_key]
+        
+        # Run inference
+        input_tensor = self.fen_to_tensor(board.fen())
+        output = self.session.run([self.output_name], {self.input_name: input_tensor})
+        
+        # Value output (tanh normalized)
+        eval_score = float(output[0][0][0]) * 400.0  # Scale to centipawns
+        
+        # Flip for black
+        if board.turn == chess.BLACK:
+            eval_score = -eval_score
+        
+        # Cache result
+        if len(self.tt_cache) < self.max_tt_size:
+            self.tt_cache[fen_key] = eval_score
+        
+        return eval_score
+    
+    def order_moves(self, board: chess.Board, moves: List[chess.Move]) -> List[chess.Move]:
+        """
+        Simple move ordering
+        1. Captures (MVV-LVA)
+        2. Checks
+        3. Other moves
+        """
+        scored_moves = []
+        
+        for move in moves:
+            score = 0
+            
+            # Captures
+            if board.is_capture(move):
+                victim = board.piece_at(move.to_square)
+                attacker = board.piece_at(move.from_square)
+                if victim and attacker:
+                    victim_val = self.PIECE_VALUES.get(victim.piece_type, 0)
+                    attacker_val = self.PIECE_VALUES.get(attacker.piece_type, 1)
+                    score = (victim_val * 10 - attacker_val) * 100
+            
+            # Promotions
+            if move.promotion == chess.QUEEN:
+                score += 9000
+            
+            # Checks
+            board.push(move)
+            if board.is_check():
+                score += 5000
+            board.pop()
+            
+            scored_moves.append((score, move))
+        
+        scored_moves.sort(key=lambda x: x[0], reverse=True)
+        return [move for _, move in scored_moves]
+    
+    def quiescence(self, board: chess.Board, alpha: float, beta: float, depth: int = 2) -> float:
+        """Quiescence search (captures only)"""
+        
+        stand_pat = self.evaluate(board)
+        
+        if stand_pat >= beta:
+            return beta
+        if alpha < stand_pat:
+            alpha = stand_pat
+        
+        if depth == 0:
+            return stand_pat
+        
+        # Only captures
+        captures = [m for m in board.legal_moves if board.is_capture(m)]
+        if not captures:
+            return stand_pat
+        
+        captures = self.order_moves(board, captures)
+        
+        for move in captures:
+            board.push(move)
+            score = -self.quiescence(board, -beta, -alpha, depth - 1)
+            board.pop()
+            
+            if score >= beta:
+                return beta
+            if score > alpha:
+                alpha = score
+        
+        return alpha
+    
+    def alpha_beta(
+        self,
+        board: chess.Board,
+        depth: int,
+        alpha: float,
+        beta: float,
+        start_time: float,
+        time_limit: float
+    ) -> Tuple[float, Optional[chess.Move]]:
+        """Alpha-beta search"""
+        
+        # Time check
+        if time.time() - start_time > time_limit:
+            return self.evaluate(board), None
+        
+        # Terminal nodes
+        if board.is_game_over():
+            if board.is_checkmate():
+                return -10000, None
+            return 0, None
+        
+        # Leaf nodes
+        if depth == 0:
+            return self.quiescence(board, alpha, beta), None
+        
+        legal_moves = list(board.legal_moves)
+        if not legal_moves:
+            return 0, None
+        
+        ordered_moves = self.order_moves(board, legal_moves)
+        
+        best_move = ordered_moves[0]
+        best_score = float('-inf')
+        
+        for move in ordered_moves:
+            board.push(move)
+            score, _ = self.alpha_beta(board, depth - 1, -beta, -alpha, start_time, time_limit)
+            score = -score
+            board.pop()
+            
+            if score > best_score:
+                best_score = score
+                best_move = move
+            
+            alpha = max(alpha, score)
+            if alpha >= beta:
+                break
+        
+        return best_score, best_move
+    
+    def get_best_move(self, fen: str, depth: int = 4, time_limit: int = 3000) -> Dict:
+        """Main search entry"""
+        
+        board = chess.Board(fen)
+        self.nodes_evaluated = 0
+        
+        time_limit_sec = time_limit / 1000.0
+        start_time = time.time()
+        
+        # Special cases
+        legal_moves = list(board.legal_moves)
+        if len(legal_moves) == 0:
+            return {'best_move': '0000', 'evaluation': 0.0, 'depth_searched': 0, 'nodes_evaluated': 0}
+        
+        if len(legal_moves) == 1:
+            return {
+                'best_move': legal_moves[0].uci(),
+                'evaluation': round(self.evaluate(board) / 100.0, 2),
+                'depth_searched': 0,
+                'nodes_evaluated': 1,
+                'time_taken': 0
+            }
+        
+        # Iterative deepening
+        best_move = legal_moves[0]
+        best_score = float('-inf')
+        
+        for current_depth in range(1, depth + 1):
+            if time.time() - start_time > time_limit_sec * 0.9:
+                break
+            
+            try:
+                score, move = self.alpha_beta(
+                    board, current_depth,
+                    float('-inf'), float('inf'),
+                    start_time, time_limit_sec
+                )
+                
+                if move:
+                    best_move = move
+                    best_score = score
+                
+            except Exception as e:
+                logger.warning(f"Search error: {e}")
+                break
+        
+        time_taken = int((time.time() - start_time) * 1000)
+        
+        return {
+            'best_move': best_move.uci(),
+            'evaluation': round(best_score / 100.0, 2),
+            'depth_searched': current_depth,
+            'nodes_evaluated': self.nodes_evaluated,
+            'time_taken': time_taken
+        }
+    
+    def validate_fen(self, fen: str) -> bool:
+        try:
+            chess.Board(fen)
+            return True
+        except:
+            return False
+    
+    def get_model_size(self) -> float:
+        return self.model_path.stat().st_size / (1024 * 1024)