Add mcvs_vs_uct.py

mcvs_vs_uct.py

@@ -0,0 +1,352 @@
+import torch
+import chess
+import numpy as np
+import os
+import time
+import threading
+import queue
+import random
+from datetime import datetime
+from chess_mcvs import (
+    Chess,
+    PolicyNetworkCNN,
+    ValueNetworkCNN,
+    HilbertOrderedZoneDatabase,
+    MCVSSearcher,
+    UCTSearcher,
+    train_networks
+)
+
+class DetailedMoveLogger:
+    """Logs detailed move generation statistics to move_log.txt"""
+    
+    def __init__(self, filepath="move_log.txt"):
+        self.filepath = filepath
+        self.file_handle = None
+        
+    def open(self):
+        self.file_handle = open(self.filepath, 'w', encoding='utf-8')
+        self._write_header()
+    
+    def _write_header(self):
+        self.file_handle.write("=" * 100 + "\n")
+        self.file_handle.write("CHESS TOURNAMENT - MOVE GENERATION LOG (ZONE-ONLY ABLATION)\n")
+        self.file_handle.write(f"Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
+        self.file_handle.write("=" * 100 + "\n\n")
+        self.file_handle.flush()
+    
+    def log_move(self, game_num, move_num, player_name, agent_type, 
+                 simulations, time_taken, move):
+        msg = (
+            f"[Game {game_num:3d}] [Move {move_num:3d}] "
+            f"Player: {player_name:5s} ({agent_type:4s}) | "
+            f"Sims: {simulations:5d} | "
+            f"Time: {time_taken:6.3f}s | "
+            f"Move: {move}\n"
+        )
+        self.file_handle.write(msg)
+        self.file_handle.flush()
+    
+    def log_game_summary(self, game_num, winner, total_moves, mcvs_stats, uct_stats):
+        self.file_handle.write("\n" + "-" * 100 + "\n")
+        self.file_handle.write(f"GAME {game_num} SUMMARY - Winner: {winner}\n")
+        self.file_handle.write(f"Total moves: {total_moves}\n\n")
+        
+        self.file_handle.write(f"MCVS (Zone-only) Statistics:\n")
+        self.file_handle.write(f"  Total simulations: {mcvs_stats['total_sims']:,}\n")
+        self.file_handle.write(f"  Avg sims/move: {mcvs_stats['avg_sims']:.1f}\n")
+        self.file_handle.write(f"  Total time: {mcvs_stats['total_time']:.2f}s\n")
+        self.file_handle.write(f"  Avg time/move: {mcvs_stats['avg_time']:.3f}s\n\n")
+        
+        self.file_handle.write(f"UCT Statistics:\n")
+        self.file_handle.write(f"  Total simulations: {uct_stats['total_sims']:,}\n")
+        self.file_handle.write(f"  Avg sims/move: {uct_stats['avg_sims']:.1f}\n")
+        self.file_handle.write(f"  Total time: {uct_stats['total_time']:.2f}s\n")
+        self.file_handle.write(f"  Avg time/move: {uct_stats['avg_time']:.3f}s\n")
+        self.file_handle.write("-" * 100 + "\n\n")
+        self.file_handle.flush()
+    
+    def close(self):
+        if self.file_handle:
+            self.file_handle.write("\n" + "=" * 100 + "\n")
+            self.file_handle.write(f"Completed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
+            self.file_handle.write("=" * 100 + "\n")
+            self.file_handle.close()
+
+class ChessTournament:
+    def __init__(self, num_games=200, time_per_move=3.0, max_moves_per_game=200, device='cpu', 
+                 output_file="chess_zone_only_results.txt", ablation_no_nets=True):
+        self.num_games = num_games
+        self.time_per_move = time_per_move
+        self.max_moves_per_game = max_moves_per_game
+        self.device = device
+        self.output_file = output_file
+        self.ablation_no_nets = ablation_no_nets
+
+        self.move_logger = DetailedMoveLogger()
+        self.move_logger.open()
+
+        self.zone_db = HilbertOrderedZoneDatabase("chess_zone_db.npz")
+
+        # ABLATION: Zone-only (no networks)
+        self.agent_mcvs = MCVSSearcher(
+            policy_net=None, value_net=None, zone_db=self.zone_db,
+            device=device, lambda_zone=1.0, use_nets=False
+        )
+        self.agent_uct = UCTSearcher()
+
+        self.report_lines = []
+
+        self.results = {
+            "MCVS": {"wins": 0, "as_white": 0, "as_black": 0, "moves": []},
+            "UCT": {"wins": 0, "as_white": 0, "as_black": 0, "moves": []},
+            "Draw": {"count": 0, "moves": []}
+        }
+
+    def log_line(self, text):
+        print(text)
+        self.report_lines.append(text)
+
+    def extract_training_and_zone_data(self, moves_made, pi_records, outcome):
+        game = Chess()
+        W_states = []
+        policies = []
+        values = []
+
+        winner_color = outcome.winner if outcome and outcome.winner is not None else None
+
+        for step, move in enumerate(moves_made):
+            W = game.get_weighted_adjacency_matrix().copy()
+            W_states.append(W)
+
+            legal_moves = game.get_legal_moves()
+            pi = np.zeros(4096, dtype=np.float32)
+
+            visits = pi_records[step]
+            if visits is not None and sum(visits.values()) > 0:
+                total = sum(visits.values())
+                for m, count in visits.items():
+                    idx = m.from_square * 64 + m.to_square
+                    pi[idx] = count / total
+            else:
+                if legal_moves:
+                    p = 1.0 / len(legal_moves)
+                    for m in legal_moves:
+                        idx = m.from_square * 64 + m.to_square
+                        pi[idx] = p
+
+            policies.append(pi)
+
+            if winner_color is None:
+                v = 0.0
+            else:
+                ptm_white = (step % 2 == 0)
+                ptm = chess.WHITE if ptm_white else chess.BLACK
+                v = 1.0 if winner_color == ptm else -1.0
+            values.append(v)
+
+            game.apply_move(move)
+
+        return W_states, policies, values
+
+    def play_game(self, agent_white, agent_black, game_idx):
+        game = Chess()
+        moves = 0
+        moves_made = []
+        pi_records = []
+        mcvs_stats = {"total_sims": 0, "total_time": 0.0, "moves": 0}
+        uct_stats = {"total_sims": 0, "total_time": 0.0, "moves": 0}
+
+        while not game.is_terminal() and moves < self.max_moves_per_game:
+            moves += 1
+            is_white = game.board.turn
+            searcher = agent_white if is_white else agent_black
+            agent_type = "MCVS" if searcher is self.agent_mcvs else "UCT"
+
+            if agent_type == "MCVS":
+                mcvs_stats["moves"] += 1
+            else:
+                uct_stats["moves"] += 1
+
+            start_t = time.time()
+            visits, sims = searcher.search_with_time_budget(game.copy(), self.time_per_move)
+            t_taken = time.time() - start_t
+
+            if agent_type == "MCVS":
+                mcvs_stats["total_sims"] += sims
+                mcvs_stats["total_time"] += t_taken
+            else:
+                uct_stats["total_sims"] += sims
+                uct_stats["total_time"] += t_taken
+
+            chosen_move = max(visits, key=visits.get) if visits else random.choice(game.get_legal_moves())
+
+            player_name = "White" if is_white else "Black"
+            self.move_logger.log_move(game_idx, moves, player_name, agent_type, sims, t_taken, chosen_move.uci())
+
+            game.apply_move(chosen_move)
+            moves_made.append(chosen_move)
+            pi_records.append(visits if agent_type == "MCVS" else None)
+
+        if mcvs_stats["moves"] > 0:
+            mcvs_stats["avg_sims"] = mcvs_stats["total_sims"] / mcvs_stats["moves"]
+            mcvs_stats["avg_time"] = mcvs_stats["total_time"] / mcvs_stats["moves"]
+        if uct_stats["moves"] > 0:
+            uct_stats["avg_sims"] = uct_stats["total_sims"] / uct_stats["moves"]
+            uct_stats["avg_time"] = uct_stats["total_time"] / uct_stats["moves"]
+
+        if moves >= self.max_moves_per_game:
+            winner = "Draw"
+            outcome = None
+            self.move_logger.log_game_summary(game_idx, "Draw (move limit)", moves, mcvs_stats, uct_stats)
+        else:
+            outcome = game.board.outcome(claim_draw=True)
+            if outcome is None or outcome.winner is None:
+                winner = "Draw"
+            elif outcome.winner == chess.WHITE:
+                winner = "MCVS" if agent_white is self.agent_mcvs else "UCT"
+            else:
+                winner = "MCVS" if agent_black is self.agent_mcvs else "UCT"
+            self.move_logger.log_game_summary(game_idx, winner, moves, mcvs_stats, uct_stats)
+
+        return winner, moves, moves_made, pi_records, outcome
+
+    def run_tournament(self):
+        start_time = time.time()
+        self.log_line("Starting 200-game Chess ablation tournament (zone guidance only vs UCT)")
+
+        for game_idx in range(1, self.num_games + 1):
+            if game_idx % 2 == 1:
+                winner, moves, moves_made, pi_records, outcome = self.play_game(self.agent_mcvs, self.agent_uct, game_idx)
+                white_name = "MCVS"
+            else:
+                winner, moves, moves_made, pi_records, outcome = self.play_game(self.agent_uct, self.agent_mcvs, game_idx)
+                white_name = "UCT"
+
+            if winner == "Draw":
+                self.results["Draw"]["count"] += 1
+                self.results["Draw"]["moves"].append(moves)
+            else:
+                self.results[winner]["wins"] += 1
+                self.results[winner]["moves"].append(moves)
+                if winner == white_name:
+                    self.results[winner]["as_white"] += 1
+                else:
+                    self.results[winner]["as_black"] += 1
+
+            W_states, policies, values = self.extract_training_and_zone_data(moves_made, pi_records, outcome)
+
+            # Sample positions for zone DB (always, even in ablation)
+            selected = [0]
+            if len(W_states) > 1:
+                selected.append(len(W_states)-1)
+                for i in range(1, len(W_states)-1):
+                    if random.random() < 0.3:
+                        selected.append(i)
+
+            # Label zone DB entries by the GAME-LEVEL ENGINE OUTCOME (which engine
+            # won the game), NOT by the per-position training value. Training
+            # targets (`values`) are still used for policy/value training, but
+            # the zone DB should represent whether this position came from a
+            # trajectory belonging to an engine win/loss/draw (see paper).
+            old_w = len(self.zone_db.winning_matrices)
+            old_l = len(self.zone_db.losing_matrices)
+            old_d = len(self.zone_db.draw_matrices)
+
+            for i in selected:
+                W = W_states[i]
+                if winner == "MCVS":
+                    # This game's winner was MCVS → label these positions as winning
+                    self.zone_db.add_winning_matrix(W)
+                elif winner == "UCT":
+                    # This game's winner was UCT → label these positions as losing
+                    self.zone_db.add_losing_matrix(W)
+                else:
+                    # Draw
+                    self.zone_db.add_draw_matrix(W)
+
+            # Sanity check: ensure we added exactly `len(selected)` matrices to
+            # the zone corresponding to the game-level outcome.
+            new_w = len(self.zone_db.winning_matrices)
+            new_l = len(self.zone_db.losing_matrices)
+            new_d = len(self.zone_db.draw_matrices)
+            if winner == "MCVS":
+                assert new_w == old_w + len(selected), f"Zone DB labeling mismatch: expected {len(selected)} winning matrices added, got {new_w - old_w}"
+            elif winner == "UCT":
+                assert new_l == old_l + len(selected), f"Zone DB labeling mismatch: expected {len(selected)} losing matrices added, got {new_l - old_l}"
+            else:
+                assert new_d == old_d + len(selected), f"Zone DB labeling mismatch: expected {len(selected)} draw matrices added, got {new_d - old_d}"
+
+            # SAVE ZONE DB AFTER EVERY GAME
+            self.zone_db.save()
+
+            # Progress logging
+            elapsed = time.time() - start_time
+            remaining = elapsed / game_idx * (self.num_games - game_idx)
+            db_size = len(self.zone_db.winning_matrices) + len(self.zone_db.losing_matrices) + len(self.zone_db.draw_matrices)
+            self.log_line(
+                f"Game {game_idx:3d}/{self.num_games} | {winner:10s} (moves={moves:3d}) | "
+                f"MCVS wins: {self.results['MCVS']['wins']:3d} | Draws: {self.results['Draw']['count']:3d} | "
+                f"DB size: {db_size:,} | ETA: {remaining/60:.1f} min"
+            )
+
+        # Final summary
+        self.log_line("\n" + "=" * 80)
+        self.log_line("ABLATION TOURNAMENT SUMMARY (Zone-only vs UCT)")
+        self.log_line("=" * 80)
+        
+        mcvs_wins = self.results["MCVS"]["wins"]
+        uct_wins = self.results["UCT"]["wins"]
+        draws = self.results["Draw"]["count"]
+        
+        self.log_line(f"\nMCVS (Zone-only λ-PUCT): {mcvs_wins} wins")
+        self.log_line(f"UCT (Standard): {uct_wins} wins")
+        self.log_line(f"Draws: {draws}")
+        self.log_line(f"Total: {self.num_games}")
+        
+        if mcvs_wins + uct_wins > 0:
+            win_rate = (mcvs_wins / (mcvs_wins + uct_wins)) * 100
+            self.log_line(f"\nMCVS Win Rate (excluding draws): {win_rate:.1f}%")
+        
+        if self.results["MCVS"]["moves"]:
+            avg_moves_mcvs = np.mean(self.results["MCVS"]["moves"])
+            self.log_line(f"Avg moves (MCVS wins): {avg_moves_mcvs:.1f}")
+        if self.results["UCT"]["moves"]:
+            avg_moves_uct = np.mean(self.results["UCT"]["moves"])
+            self.log_line(f"Avg moves (UCT wins): {avg_moves_uct:.1f}")
+        
+        self.log_line(f"\nMCVS White: {self.results['MCVS']['as_white']}, Black: {self.results['MCVS']['as_black']}")
+        self.log_line(f"UCT White: {self.results['UCT']['as_white']}, Black: {self.results['UCT']['as_black']}")
+        
+        total_db = len(self.zone_db.winning_matrices) + len(self.zone_db.losing_matrices) + len(self.zone_db.draw_matrices)
+        self.log_line(f"\nZone Database Statistics:")
+        self.log_line(f"  Winning positions: {len(self.zone_db.winning_matrices)}")
+        self.log_line(f"  Losing positions: {len(self.zone_db.losing_matrices)}")
+        self.log_line(f"  Draw positions: {len(self.zone_db.draw_matrices)}")
+        self.log_line(f"  Total positions: {total_db:,}")
+
+        total_time = time.time() - start_time
+        self.log_line("\n" + "=" * 80)
+        self.log_line(f"Tournament Duration: {total_time/60:.1f} minutes")
+        self.log_line(f"Avg time per game: {total_time/self.num_games:.1f}s")
+        self.log_line("=" * 80 + "\n")
+
+        with open(self.output_file, 'w', encoding='utf-8') as f:
+            f.write('\n'.join(self.report_lines))
+
+        self.move_logger.close()
+
+if __name__ == "__main__":
+    DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu'
+    print(f"Using device: {DEVICE}")
+
+    print("\n=== RUNNING ABLATION: Zone guidance ONLY (no neural networks) ===")
+    tournament_ablation = ChessTournament(
+        num_games=100,
+        time_per_move=1.0,
+        max_moves_per_game=200,
+        device=DEVICE,
+        output_file="chess_zone_only_results.txt",
+        ablation_no_nets=True
+    )
+    tournament_ablation.run_tournament()