Create engine/search.py

engine/search.py

@@ -1,289 +1,424 @@
 """
 Nexus-Core Search Engine
-Efficient alpha-beta with essential optimizations
-- Basic transposition table
-- Simple move ordering (MVV-LVA)
-- Quiescence search
+PVS with advanced pruning techniques
+
+Research Implementation:
+- Principal Variation Search (Marsland, 1986)
+- Null Move Pruning (Donninger, 1993)
+- Late Move Reductions (Heinz, 2000)
+- Quiescence Search (Harris, 1975)
 """
 
-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
+from typing import Optional, Tuple, List, Dict
+
+from .evaluate import NexusCoreEvaluator
+from .transposition import TranspositionTable, NodeType
+from .move_ordering import MoveOrderer
+from .time_manager import TimeManager
+from .endgame import EndgameDetector
 
 logger = logging.getLogger(__name__)
 
 
 class NexusCoreEngine:
-    """
-    Lightweight chess engine for Nexus-Core
-    Optimized for speed over strength
-    """
+    """Nexus-Core chess engine with 13.2M parameter neural network"""
+    
+    MATE_SCORE = 100000
+    MAX_PLY = 100
     
-    PIECE_VALUES = {
-        chess.PAWN: 100,
-        chess.KNIGHT: 320,
-        chess.BISHOP: 330,
-        chess.ROOK: 500,
-        chess.QUEEN: 900,
-        chess.KING: 0
-    }
+    # Pruning parameters
+    NULL_MOVE_REDUCTION = 2
+    NULL_MOVE_MIN_DEPTH = 3
+    LMR_MIN_DEPTH = 3
+    LMR_MOVE_THRESHOLD = 4
+    ASPIRATION_WINDOW = 50
     
     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
+        self.evaluator = NexusCoreEvaluator(model_path, num_threads)
+        self.tt = TranspositionTable(size_mb=128)  # 128MB TT
+        self.move_orderer = MoveOrderer()
+        self.time_manager = TimeManager()
+        self.endgame_detector = EndgameDetector()
         
         # Statistics
         self.nodes_evaluated = 0
+        self.depth_reached = 0
+        self.sel_depth = 0
+        self.principal_variation = []
         
-        logger.info("✅ Nexus-Core engine ready")
+        logger.info("🎯 Nexus-Core Engine initialized")
+        logger.info(f"   Model: {self.evaluator.get_model_size_mb():.2f} MB")
+        logger.info(f"   TT Size: 128 MB")
     
-    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)
+    def get_best_move(
+        self,
+        fen: str,
+        depth: int = 5,
+        time_limit: int = 3000
+    ) -> Dict:
+        """
+        Main search entry point
         
-        piece_to_channel = {
-            chess.PAWN: 0, chess.KNIGHT: 1, chess.BISHOP: 2,
-            chess.ROOK: 3, chess.QUEEN: 4, chess.KING: 5
-        }
+        Args:
+            fen: Position in FEN
+            depth: Max search depth
+            time_limit: Time limit in ms
         
-        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
+        Returns:
+            Dictionary with best_move and stats
+        """
         
-        return tensor
-    
-    def evaluate(self, board: chess.Board) -> float:
-        """Neural network evaluation"""
-        self.nodes_evaluated += 1
+        board = chess.Board(fen)
         
-        # Check cache (simple FEN-based)
-        fen_key = board.fen().split(' ')[0]
-        if fen_key in self.tt_cache:
-            return self.tt_cache[fen_key]
+        # Reset stats
+        self.nodes_evaluated = 0
+        self.depth_reached = 0
+        self.sel_depth = 0
+        self.principal_variation = []
+        
+        # Time management
+        time_limit_sec = time_limit / 1000.0
+        self.time_manager.start_search(time_limit_sec, time_limit_sec)
         
-        # Run inference
-        input_tensor = self.fen_to_tensor(board.fen())
-        output = self.session.run([self.output_name], {self.input_name: input_tensor})
+        # Age history
+        self.move_orderer.age_history(0.95)
+        self.tt.increment_age()
         
-        # Value output (tanh normalized)
-        eval_score = float(output[0][0][0]) * 400.0  # Scale to centipawns
+        # Special cases
+        legal_moves = list(board.legal_moves)
         
-        # Flip for black
-        if board.turn == chess.BLACK:
-            eval_score = -eval_score
+        if len(legal_moves) == 0:
+            return self._no_legal_moves_result()
         
-        # Cache result
-        if len(self.tt_cache) < self.max_tt_size:
-            self.tt_cache[fen_key] = eval_score
+        if len(legal_moves) == 1:
+            return self._single_move_result(board, legal_moves[0])
         
-        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 = []
+        # Iterative deepening
+        best_move = legal_moves[0]
+        best_score = float('-inf')
+        alpha = float('-inf')
+        beta = float('inf')
         
-        for move in moves:
-            score = 0
+        for current_depth in range(1, depth + 1):
+            if self.time_manager.should_stop(current_depth):
+                break
             
-            # 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
+            # Aspiration windows for depth >= 4
+            if current_depth >= 4 and abs(best_score) < self.MATE_SCORE - 1000:
+                alpha = best_score - self.ASPIRATION_WINDOW
+                beta = best_score + self.ASPIRATION_WINDOW
+            else:
+                alpha = float('-inf')
+                beta = float('inf')
             
-            # Promotions
-            if move.promotion == chess.QUEEN:
-                score += 9000
+            # Search
+            score, move, pv = self._search_root(
+                board, current_depth, alpha, beta
+            )
             
-            # Checks
-            board.push(move)
-            if board.is_check():
-                score += 5000
-            board.pop()
+            # Handle aspiration failures
+            if score <= alpha or score >= beta:
+                score, move, pv = self._search_root(
+                    board, current_depth, float('-inf'), float('inf')
+                )
             
-            scored_moves.append((score, move))
+            # Update best
+            if move:
+                best_move = move
+                best_score = score
+                self.depth_reached = current_depth
+                self.principal_variation = pv
+                
+                logger.info(
+                    f"Depth {current_depth}: {move.uci()} "
+                    f"({score:+.2f}) | Nodes: {self.nodes_evaluated}"
+                )
         
-        scored_moves.sort(key=lambda x: x[0], reverse=True)
-        return [move for _, move in scored_moves]
+        return {
+            'best_move': best_move.uci(),
+            'evaluation': round(best_score / 100.0, 2),
+            'depth_searched': self.depth_reached,
+            'seldepth': self.sel_depth,
+            'nodes_evaluated': self.nodes_evaluated,
+            'time_taken': int(self.time_manager.elapsed() * 1000),
+            'pv': [m.uci() for m in self.principal_variation],
+            'nps': int(self.nodes_evaluated / max(self.time_manager.elapsed(), 0.001)),
+            'tt_stats': self.tt.get_stats(),
+            'move_ordering_stats': self.move_orderer.get_stats()
+        }
     
-    def quiescence(self, board: chess.Board, alpha: float, beta: float, depth: int = 2) -> float:
-        """Quiescence search (captures only)"""
-        
-        stand_pat = self.evaluate(board)
+    def _search_root(
+        self,
+        board: chess.Board,
+        depth: int,
+        alpha: float,
+        beta: float
+    ) -> Tuple[float, Optional[chess.Move], List[chess.Move]]:
+        """Root node search"""
         
-        if stand_pat >= beta:
-            return beta
-        if alpha < stand_pat:
-            alpha = stand_pat
+        legal_moves = list(board.legal_moves)
         
-        if depth == 0:
-            return stand_pat
+        # TT probe
+        zobrist_key = self.tt.compute_zobrist_key(board)
+        tt_result = self.tt.probe(zobrist_key, depth, alpha, beta)
+        tt_move = tt_result[1] if tt_result else None
         
-        # Only captures
-        captures = [m for m in board.legal_moves if board.is_capture(m)]
-        if not captures:
-            return stand_pat
+        # Order moves
+        ordered_moves = self.move_orderer.order_moves(
+            board, legal_moves, depth, tt_move
+        )
         
-        captures = self.order_moves(board, captures)
+        best_move = ordered_moves[0]
+        best_score = float('-inf')
+        best_pv = []
         
-        for move in captures:
+        for i, move in enumerate(ordered_moves):
             board.push(move)
-            score = -self.quiescence(board, -beta, -alpha, depth - 1)
+            
+            if i == 0:
+                score, pv = self._pvs(
+                    board, depth - 1, -beta, -alpha, True
+                )
+                score = -score
+            else:
+                score, _ = self._pvs(
+                    board, depth - 1, -alpha - 1, -alpha, False
+                )
+                score = -score
+                
+                if alpha < score < beta:
+                    score, pv = self._pvs(
+                        board, depth - 1, -beta, -alpha, True
+                    )
+                    score = -score
+                else:
+                    pv = []
+            
             board.pop()
             
-            if score >= beta:
-                return beta
+            if score > best_score:
+                best_score = score
+                best_move = move
+                best_pv = [move] + pv
+            
             if score > alpha:
                 alpha = score
+            
+            if self.time_manager.should_stop(depth):
+                break
         
-        return alpha
+        # Store in TT
+        self.tt.store(zobrist_key, depth, best_score, NodeType.EXACT, best_move)
+        
+        return best_score, best_move, best_pv
     
-    def alpha_beta(
+    def _pvs(
         self,
         board: chess.Board,
         depth: int,
         alpha: float,
         beta: float,
-        start_time: float,
-        time_limit: float
-    ) -> Tuple[float, Optional[chess.Move]]:
-        """Alpha-beta search"""
+        do_null: bool
+    ) -> Tuple[float, List[chess.Move]]:
+        """Principal Variation Search"""
+        
+        self.sel_depth = max(self.sel_depth, self.MAX_PLY - depth)
+        
+        # Mate distance pruning
+        alpha = max(alpha, -self.MATE_SCORE + (self.MAX_PLY - depth))
+        beta = min(beta, self.MATE_SCORE - (self.MAX_PLY - depth) - 1)
+        if alpha >= beta:
+            return alpha, []
+        
+        # Draw detection
+        if board.is_repetition(2) or board.is_fifty_moves():
+            return 0, []
+        
+        # TT probe
+        zobrist_key = self.tt.compute_zobrist_key(board)
+        tt_result = self.tt.probe(zobrist_key, depth, alpha, beta)
         
-        # Time check
-        if time.time() - start_time > time_limit:
-            return self.evaluate(board), None
+        if tt_result and tt_result[0] is not None:
+            return tt_result[0], []
         
-        # Terminal nodes
-        if board.is_game_over():
-            if board.is_checkmate():
-                return -10000, None
-            return 0, None
+        tt_move = tt_result[1] if tt_result else None
         
-        # Leaf nodes
-        if depth == 0:
-            return self.quiescence(board, alpha, beta), None
+        # Quiescence search
+        if depth <= 0:
+            return self._quiescence(board, alpha, beta, 0), []
+        
+        # Null move pruning
+        if (do_null and 
+            depth >= self.NULL_MOVE_MIN_DEPTH and
+            not board.is_check() and
+            self._has_non_pawn_material(board)):
+            
+            board.push(chess.Move.null())
+            score, _ = self._pvs(
+                board, depth - 1 - self.NULL_MOVE_REDUCTION,
+                -beta, -beta + 1, False
+            )
+            score = -score
+            board.pop()
+            
+            if score >= beta:
+                return beta, []
         
+        # Generate moves
         legal_moves = list(board.legal_moves)
         if not legal_moves:
-            return 0, None
+            if board.is_check():
+                return -self.MATE_SCORE + (self.MAX_PLY - depth), []
+            return 0, []
         
-        ordered_moves = self.order_moves(board, legal_moves)
+        ordered_moves = self.move_orderer.order_moves(
+            board, legal_moves, depth, tt_move
+        )
         
-        best_move = ordered_moves[0]
+        # Main search
         best_score = float('-inf')
+        best_pv = []
+        node_type = NodeType.UPPER_BOUND
+        moves_searched = 0
         
         for move in ordered_moves:
             board.push(move)
-            score, _ = self.alpha_beta(board, depth - 1, -beta, -alpha, start_time, time_limit)
-            score = -score
+            
+            # Late move reductions
+            reduction = 0
+            if (moves_searched >= self.LMR_MOVE_THRESHOLD and
+                depth >= self.LMR_MIN_DEPTH and
+                not board.is_check() and
+                not board.is_capture(board.peek())):
+                reduction = 1
+            
+            # PVS
+            if moves_searched == 0:
+                score, pv = self._pvs(
+                    board, depth - 1, -beta, -alpha, True
+                )
+                score = -score
+            else:
+                score, _ = self._pvs(
+                    board, depth - 1 - reduction, -alpha - 1, -alpha, True
+                )
+                score = -score
+                
+                if alpha < score < beta:
+                    score, pv = self._pvs(
+                        board, depth - 1, -beta, -alpha, True
+                    )
+                    score = -score
+                else:
+                    pv = []
+            
             board.pop()
+            moves_searched += 1
             
             if score > best_score:
                 best_score = score
-                best_move = move
-            
-            alpha = max(alpha, score)
-            if alpha >= beta:
-                break
-        
-        return best_score, best_move
+                best_pv = [move] + pv
+                
+                if score > alpha:
+                    alpha = score
+                    node_type = NodeType.EXACT
+                    
+                    if not board.is_capture(move):
+                        self.move_orderer.update_history(move, depth, True)
+                        self.move_orderer.update_killer_move(move, depth)
+                    
+                    if score >= beta:
+                        node_type = NodeType.LOWER_BOUND
+                        break
+        
+        self.tt.store(zobrist_key, depth, best_score, node_type, best_pv[0] if best_pv else None)
+        
+        return best_score, best_pv
     
-    def get_best_move(self, fen: str, depth: int = 4, time_limit: int = 3000) -> Dict:
-        """Main search entry"""
+    def _quiescence(
+        self,
+        board: chess.Board,
+        alpha: float,
+        beta: float,
+        qs_depth: int
+    ) -> float:
+        """Quiescence search"""
         
-        board = chess.Board(fen)
-        self.nodes_evaluated = 0
+        self.nodes_evaluated += 1
         
-        time_limit_sec = time_limit / 1000.0
-        start_time = time.time()
+        # Stand-pat
+        stand_pat = self.evaluator.evaluate_hybrid(board)
+        stand_pat = self.endgame_detector.adjust_evaluation(board, stand_pat)
         
-        # 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 stand_pat >= beta:
+            return beta
+        if alpha < stand_pat:
+            alpha = stand_pat
         
-        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
-            }
+        # Depth limit
+        if qs_depth >= 8:
+            return stand_pat
         
-        # Iterative deepening
-        best_move = legal_moves[0]
-        best_score = float('-inf')
+        # Tactical moves
+        tactical_moves = [
+            move for move in board.legal_moves
+            if board.is_capture(move) or board.gives_check(move)
+        ]
         
-        for current_depth in range(1, depth + 1):
-            if time.time() - start_time > time_limit_sec * 0.9:
-                break
+        if not tactical_moves:
+            return stand_pat
+        
+        tactical_moves = self.move_orderer.order_moves(board, tactical_moves, 0)
+        
+        for move in tactical_moves:
+            board.push(move)
+            score = -self._quiescence(board, -beta, -alpha, qs_depth + 1)
+            board.pop()
             
-            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
+            if score >= beta:
+                return beta
+            if score > alpha:
+                alpha = score
         
-        time_taken = int((time.time() - start_time) * 1000)
+        return alpha
+    
+    def _has_non_pawn_material(self, board: chess.Board) -> bool:
+        """Check for non-pawn material"""
+        for piece_type in [chess.KNIGHT, chess.BISHOP, chess.ROOK, chess.QUEEN]:
+            if board.pieces(piece_type, board.turn):
+                return True
+        return False
+    
+    def _no_legal_moves_result(self) -> Dict:
+        """No legal moves result"""
+        return {
+            'best_move': '0000',
+            'evaluation': 0.0,
+            'depth_searched': 0,
+            'nodes_evaluated': 0,
+            'time_taken': 0
+        }
+    
+    def _single_move_result(self, board: chess.Board, move: chess.Move) -> Dict:
+        """Single move result"""
+        eval_score = self.evaluator.evaluate_hybrid(board)
         
         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
+            'best_move': move.uci(),
+            'evaluation': round(eval_score / 100.0, 2),
+            'depth_searched': 0,
+            'nodes_evaluated': 1,
+            'time_taken': 0,
+            'pv': [move.uci()]
         }
     
     def validate_fen(self, fen: str) -> bool:
+        """Validate FEN"""
         try:
             chess.Board(fen)
             return True
@@ -291,4 +426,5 @@ class NexusCoreEngine:
             return False
     
     def get_model_size(self) -> float:
-        return self.model_path.stat().st_size / (1024 * 1024)
+        """Get model size"""
+        return self.evaluator.get_model_size_mb()