mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-06 12:23:06 +02:00
feat: automatic time budgeding
This commit is contained in:
parent
0e09b6e211
commit
7ff1c91176
@ -11,7 +11,7 @@ class RandomEngine:
|
||||
Keeps the same name for compatibility, but no longer picks purely random moves.
|
||||
"""
|
||||
|
||||
def __init__(self, depth: int = 100, max_time_sec: float = 5):
|
||||
def __init__(self, depth: int = 100, max_time_sec: float = 20):
|
||||
self.depth = depth
|
||||
self.max_time_sec = max_time_sec
|
||||
|
||||
@ -25,15 +25,42 @@ class RandomEngine:
|
||||
chess.KING: 0,
|
||||
}
|
||||
|
||||
def choose_move(self, board: chess.Board) -> Optional[chess.Move]:
|
||||
# Tiny hand-crafted opening book (UCIs); used only for the first few plies
|
||||
# Keys are tuples of UCI moves played so far from the starting position
|
||||
self.opening_book: dict[tuple[str, ...], list[str]] = {
|
||||
# As White (start position)
|
||||
tuple(): ["e2e4", "d2d4", "c2c4", "g1f3"],
|
||||
# As Black after 1.e4
|
||||
("e2e4",): ["e7e5", "c7c5", "e7e6", "c7c6", "g8f6", "d7d5"],
|
||||
# As Black after 1.d4
|
||||
("d2d4",): ["d7d5", "g8f6", "e7e6", "c7c5", "c7c6"],
|
||||
# As Black after 1.c4
|
||||
("c2c4",): ["e7e5", "g8f6", "c7c5", "e7e6"],
|
||||
# As Black after 1.Nf3
|
||||
("g1f3",): ["g8f6", "d7d5", "c7c5", "e7e6"],
|
||||
# A couple continuations to avoid silly early queen/rook moves
|
||||
("e2e4", "e7e5"): ["g1f3", "f1c4", "f1b5", "d2d4"],
|
||||
("e2e4", "c7c5"): ["g1f3", "d2d4", "c2c3", "b1c3"],
|
||||
("d2d4", "d7d5"): ["c2c4", "g1f3", "e2e3"],
|
||||
("d2d4", "g8f6"): ["c2c4", "g1f3", "e2e3"],
|
||||
}
|
||||
|
||||
def choose_move(self, board: chess.Board, time_budget_sec: Optional[float] = None) -> Optional[chess.Move]:
|
||||
start = time.time()
|
||||
best_move: Optional[chess.Move] = None
|
||||
# Set a per-move deadline used throughout search
|
||||
time_limit = time_budget_sec if time_budget_sec is not None else self.max_time_sec
|
||||
self._deadline = start + max(0.01, time_limit)
|
||||
best_score = -float("inf") if board.turn else float("inf")
|
||||
|
||||
# Opening book shortcut (very early only)
|
||||
book_mv = self._opening_book_move(board)
|
||||
if book_mv is not None:
|
||||
return book_mv
|
||||
|
||||
# Iterative deepening up to depth or time limit
|
||||
for d in range(1, self.depth + 1):
|
||||
elapsed = time.time() - start
|
||||
if elapsed >= self.max_time_sec:
|
||||
if time.time() >= self._deadline:
|
||||
break
|
||||
score, move = self._search_root(board, d, start)
|
||||
if move is not None:
|
||||
@ -45,19 +72,30 @@ class RandomEngine:
|
||||
return random.choice(moves) if moves else None
|
||||
return best_move
|
||||
|
||||
def choose_move_with_explanation(self, board: chess.Board) -> Tuple[Optional[chess.Move], str]:
|
||||
def choose_move_with_explanation(self, board: chess.Board, time_budget_sec: Optional[float] = None) -> Tuple[Optional[chess.Move], str]:
|
||||
"""Return the chosen move and a human-readable explanation of top candidates.
|
||||
|
||||
The explanation lists top candidates with scores and quick annotations.
|
||||
"""
|
||||
start = time.time()
|
||||
# Set a per-move deadline used throughout search
|
||||
time_limit = time_budget_sec if time_budget_sec is not None else self.max_time_sec
|
||||
self._deadline = start + max(0.01, time_limit)
|
||||
depth_used = 0
|
||||
best_move: Optional[chess.Move] = None
|
||||
scores: list[Tuple[chess.Move, float]] = []
|
||||
|
||||
# Opening book shortcut
|
||||
book_mv = self._opening_book_move(board)
|
||||
if book_mv is not None:
|
||||
try:
|
||||
return book_mv, f"opening-book: {board.san(book_mv)} ({book_mv.uci()})"
|
||||
except Exception:
|
||||
return book_mv, f"opening-book: {book_mv.uci()}"
|
||||
|
||||
# Analyze all legal moves at the root with alpha-beta to given depth/time
|
||||
for d in range(1, self.depth + 1):
|
||||
if time.time() - start >= self.max_time_sec:
|
||||
if time.time() >= self._deadline:
|
||||
break
|
||||
depth_used = d
|
||||
scores = self._analyze_root(board, d, start)
|
||||
@ -73,27 +111,42 @@ class RandomEngine:
|
||||
def annotate(m: chess.Move) -> str:
|
||||
tags = []
|
||||
if board.is_capture(m):
|
||||
tags.append("x")
|
||||
tags.append("capture")
|
||||
if m.promotion:
|
||||
tags.append(f"={chess.piece_symbol(m.promotion).upper()}")
|
||||
tags.append(f"promotes={chess.piece_symbol(m.promotion).upper()}")
|
||||
try:
|
||||
if board.gives_check(m):
|
||||
tags.append("+")
|
||||
tags.append("check")
|
||||
except Exception:
|
||||
pass
|
||||
return "".join(tags)
|
||||
if board.is_castling(m):
|
||||
tags.append("castle")
|
||||
# Centralization
|
||||
center = {chess.D4, chess.E4, chess.D5, chess.E5}
|
||||
if m.to_square in center:
|
||||
tags.append("center")
|
||||
# Development: minor piece leaves back rank
|
||||
if board.piece_at(m.from_square) and board.piece_at(m.from_square).piece_type in (chess.KNIGHT, chess.BISHOP):
|
||||
if chess.square_rank(m.from_square) in (0, 7):
|
||||
tags.append("develops")
|
||||
# Rook to (semi-)open file
|
||||
if board.piece_at(m.from_square) and board.piece_at(m.from_square).piece_type == chess.ROOK:
|
||||
file_idx = chess.square_file(m.to_square)
|
||||
if self._is_open_file(board, file_idx):
|
||||
tags.append("open-file")
|
||||
return ",".join(tags)
|
||||
|
||||
top = scores[:5]
|
||||
best_cp = top[0][1]
|
||||
lines = [
|
||||
f"depth={depth_used} time={time.time()-start:.2f}s candidates={len(scores)}",
|
||||
f"best {board.san(top[0][0])} ({top[0][0].uci()}) score={best_cp:.1f}",
|
||||
f"best {board.san(top[0][0])} ({top[0][0].uci()}) score={best_cp:.1f} reasons=[{annotate(top[0][0])}]",
|
||||
]
|
||||
if len(top) > 1:
|
||||
lines.append("alternatives:")
|
||||
for mv, sc in top[1:]:
|
||||
delta = sc - best_cp
|
||||
lines.append(f" {board.san(mv)} ({mv.uci()}) score={sc:.1f} delta={delta:+.1f} {annotate(mv)}")
|
||||
lines.append(f" {board.san(mv)} ({mv.uci()}) score={sc:.1f} delta={delta:+.1f} reasons=[{annotate(mv)}]")
|
||||
|
||||
return best_move, "\n".join(lines)
|
||||
|
||||
@ -102,7 +155,7 @@ class RandomEngine:
|
||||
beta = float("inf")
|
||||
scored: list[Tuple[chess.Move, float]] = []
|
||||
for move in self._ordered_moves(board):
|
||||
if time.time() - start >= self.max_time_sec:
|
||||
if time.time() >= self._deadline:
|
||||
break
|
||||
board.push(move)
|
||||
score = -self._alphabeta(board, depth - 1, -beta, -alpha, start)
|
||||
@ -123,7 +176,7 @@ class RandomEngine:
|
||||
|
||||
moves = self._ordered_moves(board)
|
||||
for move in moves:
|
||||
if time.time() - start >= self.max_time_sec:
|
||||
if time.time() >= self._deadline:
|
||||
break
|
||||
board.push(move)
|
||||
score = -self._alphabeta(board, depth - 1, -beta, -alpha, start)
|
||||
@ -139,12 +192,14 @@ class RandomEngine:
|
||||
|
||||
def _alphabeta(self, board: chess.Board, depth: int, alpha: float, beta: float, start: float) -> float:
|
||||
# Time cutoff
|
||||
if time.time() - start >= self.max_time_sec:
|
||||
if time.time() >= self._deadline:
|
||||
return self._evaluate(board)
|
||||
|
||||
# Terminal nodes
|
||||
if depth == 0 or board.is_game_over():
|
||||
if board.is_game_over():
|
||||
return self._evaluate(board)
|
||||
if depth == 0:
|
||||
return self._quiescence(board, alpha, beta, start)
|
||||
|
||||
best = -float("inf")
|
||||
for move in self._ordered_moves(board):
|
||||
@ -159,6 +214,29 @@ class RandomEngine:
|
||||
break
|
||||
return best
|
||||
|
||||
def _quiescence(self, board: chess.Board, alpha: float, beta: float, start: float) -> float:
|
||||
# Stand-pat
|
||||
stand_pat = self._evaluate(board)
|
||||
if stand_pat >= beta:
|
||||
return beta
|
||||
if stand_pat > alpha:
|
||||
alpha = stand_pat
|
||||
|
||||
# Explore captures to avoid horizon effect
|
||||
for move in self._ordered_moves(board):
|
||||
if time.time() >= self._deadline:
|
||||
break
|
||||
if not board.is_capture(move) and not move.promotion:
|
||||
continue
|
||||
board.push(move)
|
||||
score = -self._quiescence(board, -beta, -alpha, start)
|
||||
board.pop()
|
||||
if score >= beta:
|
||||
return beta
|
||||
if score > alpha:
|
||||
alpha = score
|
||||
return alpha
|
||||
|
||||
def _ordered_moves(self, board: chess.Board):
|
||||
# Simple move ordering: captures/promotions first, then checks
|
||||
def score_move(m: chess.Move) -> int:
|
||||
@ -172,6 +250,58 @@ class RandomEngine:
|
||||
s += 100
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Early-game heuristics to avoid silly shuffles
|
||||
early = self._is_early_game(board)
|
||||
piece = board.piece_at(m.from_square)
|
||||
if piece:
|
||||
# Strongly prefer castling when legal
|
||||
if board.is_castling(m):
|
||||
s += 500
|
||||
# Prefer developing minors from back rank
|
||||
if piece.piece_type in (chess.KNIGHT, chess.BISHOP):
|
||||
if chess.square_rank(m.from_square) in (0, 7) and not board.is_capture(m):
|
||||
s += 80
|
||||
# Penalize knights to the rim early (unless it's a capture or check)
|
||||
if early and piece.piece_type == chess.KNIGHT:
|
||||
to_file = chess.square_file(m.to_square)
|
||||
if to_file in (0, 7) and not board.is_capture(m):
|
||||
s -= 120
|
||||
# Discourage king walks when not castling
|
||||
if piece.piece_type == chess.KING and early and not board.is_castling(m):
|
||||
s -= 400
|
||||
# Mildly discourage rook moves very early if both knights and bishops are undeveloped
|
||||
if piece.piece_type == chess.ROOK and early and self._most_minors_undeveloped(board, piece.color):
|
||||
s -= 120
|
||||
# Discourage early queen sorties unless forcing
|
||||
if piece.piece_type == chess.QUEEN and early and not board.is_capture(m):
|
||||
try:
|
||||
gives_check = board.gives_check(m)
|
||||
except Exception:
|
||||
gives_check = False
|
||||
if not gives_check:
|
||||
s -= 90
|
||||
# Discourage moving the same piece twice very early if not forcing
|
||||
if early and not board.is_capture(m) and not board.is_castling(m):
|
||||
if not self._is_start_square(piece.piece_type, piece.color, m.from_square):
|
||||
# unless stepping into center
|
||||
to_center = chess.square_file(m.to_square) in (3, 4) and chess.square_rank(m.to_square) in (2, 3, 4, 5)
|
||||
if not to_center:
|
||||
s -= 60
|
||||
# Penalize weakening f-pawn pushes early
|
||||
if piece.piece_type == chess.PAWN and early and not board.is_capture(m):
|
||||
from_file = chess.square_file(m.from_square)
|
||||
from_rank = chess.square_rank(m.from_square)
|
||||
to_rank = chess.square_rank(m.to_square)
|
||||
# f2->f3 or f7->f6
|
||||
if from_file == 5:
|
||||
if piece.color == chess.WHITE and from_rank == 1 and to_rank == 2:
|
||||
s -= 120
|
||||
if piece.color == chess.BLACK and from_rank == 6 and to_rank == 5:
|
||||
s -= 120
|
||||
# small bonus for central pawn advances
|
||||
if chess.square_file(m.to_square) in (3, 4):
|
||||
s += 40
|
||||
return s
|
||||
|
||||
moves = list(board.legal_moves)
|
||||
@ -179,23 +309,196 @@ class RandomEngine:
|
||||
return moves
|
||||
|
||||
def _evaluate(self, board: chess.Board) -> float:
|
||||
# Game end conditions
|
||||
# Terminal
|
||||
if board.is_checkmate():
|
||||
return -100000 if board.turn else 100000 # side-to-move is mated
|
||||
# If it's our turn and we're checkmated, that's bad for us
|
||||
return -100000
|
||||
if board.is_stalemate() or board.is_insufficient_material() or board.can_claim_draw():
|
||||
return 0
|
||||
|
||||
# Material
|
||||
# Base material (white minus black)
|
||||
material = 0
|
||||
for square, piece in board.piece_map().items():
|
||||
val = self.piece_values[piece.piece_type]
|
||||
material += val if piece.color == chess.WHITE else -val
|
||||
piece_map = board.piece_map()
|
||||
for sq, pc in piece_map.items():
|
||||
val = self.piece_values[pc.piece_type]
|
||||
material += val if pc.color == chess.WHITE else -val
|
||||
|
||||
# Mobility (only side to move for speed; acts as tempo bonus)
|
||||
mobility = 5 * sum(1 for _ in board.legal_moves)
|
||||
if board.turn:
|
||||
mobility_term = mobility
|
||||
# Doubled pawns penalty
|
||||
dp_pen = self._doubled_pawns_penalty(board)
|
||||
|
||||
# Mobility (white - black) with small weight
|
||||
mob_w, mob_b = self._mobility(board)
|
||||
mobility_term = (mob_w - mob_b) * 1.0
|
||||
|
||||
# Centralization: reward pieces in the center (white - black)
|
||||
center = {chess.C3, chess.D3, chess.E3, chess.F3, chess.C4, chess.D4, chess.E4, chess.F4,
|
||||
chess.C5, chess.D5, chess.E5, chess.F5, chess.C6, chess.D6, chess.E6, chess.F6}
|
||||
center_score = 0
|
||||
for sq, pc in piece_map.items():
|
||||
if sq in center:
|
||||
w = 10 if pc.piece_type in (chess.KNIGHT, chess.BISHOP) else 5
|
||||
center_score += w if pc.color == chess.WHITE else -w
|
||||
|
||||
# Rooks on open files
|
||||
rook_file_bonus = 0
|
||||
for sq, pc in piece_map.items():
|
||||
if pc.piece_type == chess.ROOK:
|
||||
file_idx = chess.square_file(sq)
|
||||
if self._is_open_file(board, file_idx):
|
||||
rook_file_bonus += 15 if pc.color == chess.WHITE else -15
|
||||
|
||||
# King safety: prefer castled in middlegame (queens/rooks present)
|
||||
safety = 0
|
||||
heavy_pieces = sum(1 for p in piece_map.values() if p.piece_type in (chess.QUEEN, chess.ROOK))
|
||||
if heavy_pieces >= 3:
|
||||
wk_sq = board.king(chess.WHITE)
|
||||
bk_sq = board.king(chess.BLACK)
|
||||
safety += self._king_safety_bonus(wk_sq, chess.WHITE)
|
||||
safety -= self._king_safety_bonus(bk_sq, chess.BLACK)
|
||||
# Penalize wandering kings early if not castled squares
|
||||
if self._is_early_game(board):
|
||||
if wk_sq not in (chess.E1, chess.G1, chess.C1):
|
||||
safety -= 40
|
||||
if bk_sq not in (chess.E8, chess.G8, chess.C8):
|
||||
safety += 40
|
||||
|
||||
# Piece-square tendencies (small)
|
||||
pst = self._pst_score(board)
|
||||
|
||||
# Aggregate white-centric score then convert to side-to-move via negamax
|
||||
white_score = material - dp_pen + mobility_term + center_score + rook_file_bonus + safety + pst
|
||||
return white_score if board.turn == chess.WHITE else -white_score
|
||||
|
||||
def _opening_book_move(self, board: chess.Board) -> Optional[chess.Move]:
|
||||
# Only use book for the first few plies and only from starting positions
|
||||
if board.move_stack is None:
|
||||
return None
|
||||
if board.fullmove_number > 10:
|
||||
return None
|
||||
# Build UCI history from the start position
|
||||
hist = tuple(m.uci() for m in board.move_stack)
|
||||
# Try exact key; also try from a truncated start if someone inserted off-book early
|
||||
for klen in range(len(hist), -1, -1):
|
||||
key = hist[:klen]
|
||||
if key in self.opening_book:
|
||||
candidates = self.opening_book[key]
|
||||
# Filter to legal moves only
|
||||
legals = {m.uci(): m for m in board.legal_moves}
|
||||
legal_ucis = [u for u in candidates if u in legals]
|
||||
if legal_ucis:
|
||||
# Choose the first candidate to be stable; could randomize if desired
|
||||
return legals[legal_ucis[0]]
|
||||
return None
|
||||
|
||||
def _is_start_square(self, piece_type: chess.PieceType, color: chess.Color, sq: int) -> bool:
|
||||
file_idx = chess.square_file(sq)
|
||||
rank_idx = chess.square_rank(sq)
|
||||
if piece_type == chess.KING:
|
||||
return (file_idx, rank_idx) == ((4, 0) if color == chess.WHITE else (4, 7))
|
||||
if piece_type == chess.QUEEN:
|
||||
return (file_idx, rank_idx) == ((3, 0) if color == chess.WHITE else (3, 7))
|
||||
if piece_type == chess.ROOK:
|
||||
return (file_idx, rank_idx) in ({(0, 0), (7, 0)} if color == chess.WHITE else {(0, 7), (7, 7)})
|
||||
if piece_type == chess.BISHOP:
|
||||
return (file_idx, rank_idx) in ({(2, 0), (5, 0)} if color == chess.WHITE else {(2, 7), (5, 7)})
|
||||
if piece_type == chess.KNIGHT:
|
||||
return (file_idx, rank_idx) in ({(1, 0), (6, 0)} if color == chess.WHITE else {(1, 7), (6, 7)})
|
||||
if piece_type == chess.PAWN:
|
||||
return rank_idx == (1 if color == chess.WHITE else 6)
|
||||
return False
|
||||
|
||||
def _pst_score(self, board: chess.Board) -> int:
|
||||
score = 0
|
||||
for sq, pc in board.piece_map().items():
|
||||
file_idx = chess.square_file(sq)
|
||||
rank_idx = chess.square_rank(sq)
|
||||
sign = 1 if pc.color == chess.WHITE else -1
|
||||
if pc.piece_type == chess.KNIGHT:
|
||||
# Knights: center good, rim bad
|
||||
if file_idx in (0, 7):
|
||||
score -= 20 * sign
|
||||
elif file_idx in (1, 6):
|
||||
score -= 10 * sign
|
||||
if rank_idx in (0, 7):
|
||||
score -= 10 * sign
|
||||
if (file_idx, rank_idx) in {(2, 2), (3, 2), (4, 2), (5, 2), (2, 3), (3, 3), (4, 3), (5, 3)}:
|
||||
score += 15 * sign
|
||||
elif pc.piece_type == chess.BISHOP:
|
||||
# Bishops: prefer long diagonals and central ranks
|
||||
if rank_idx in (2, 3, 4, 5):
|
||||
score += 5 * sign
|
||||
elif pc.piece_type == chess.PAWN:
|
||||
# Central pawns advanced are nice
|
||||
if file_idx in (3, 4):
|
||||
score += rank_idx * 1 * sign if pc.color == chess.WHITE else (7 - rank_idx) * 1 * sign
|
||||
return score
|
||||
|
||||
def _is_early_game(self, board: chess.Board) -> bool:
|
||||
# Quick heuristic for opening/middlegame
|
||||
heavy_pieces = sum(1 for p in board.piece_map().values() if p.piece_type in (chess.QUEEN, chess.ROOK))
|
||||
return heavy_pieces >= 3 and board.fullmove_number < 15
|
||||
|
||||
def _most_minors_undeveloped(self, board: chess.Board, color: chess.Color) -> bool:
|
||||
# True if 3 or 4 minors still on back rank starting squares
|
||||
if color == chess.WHITE:
|
||||
starts = [chess.B1, chess.G1, chess.C1, chess.F1]
|
||||
else:
|
||||
mobility_term = -mobility
|
||||
starts = [chess.B8, chess.G8, chess.C8, chess.F8]
|
||||
cnt = 0
|
||||
for sq in starts:
|
||||
pc = board.piece_at(sq)
|
||||
if pc and pc.color == color and pc.piece_type in (chess.KNIGHT, chess.BISHOP):
|
||||
cnt += 1
|
||||
return cnt >= 3
|
||||
|
||||
return material + mobility_term
|
||||
def _mobility(self, board: chess.Board) -> Tuple[int, int]:
|
||||
# Count legal moves for both sides using copies
|
||||
w_board = board if board.turn == chess.WHITE else board.copy(stack=False)
|
||||
if w_board.turn != chess.WHITE:
|
||||
w_board.turn = chess.WHITE
|
||||
b_board = board if board.turn == chess.BLACK else board.copy(stack=False)
|
||||
if b_board.turn != chess.BLACK:
|
||||
b_board.turn = chess.BLACK
|
||||
return sum(1 for _ in w_board.legal_moves), sum(1 for _ in b_board.legal_moves)
|
||||
|
||||
def _is_open_file(self, board: chess.Board, file_idx: int) -> bool:
|
||||
# True if no pawns on this file (either color)
|
||||
for rank in range(8):
|
||||
sq = chess.square(file_idx, rank)
|
||||
pc = board.piece_at(sq)
|
||||
if pc and pc.piece_type == chess.PAWN:
|
||||
return False
|
||||
return True
|
||||
|
||||
def _doubled_pawns_penalty(self, board: chess.Board) -> int:
|
||||
# Penalty in centipawns for doubled pawns (per extra pawn on a file)
|
||||
penalty = 0
|
||||
for color in (chess.WHITE, chess.BLACK):
|
||||
for file_idx in range(8):
|
||||
cnt = 0
|
||||
for rank in range(8):
|
||||
sq = chess.square(file_idx, rank)
|
||||
pc = board.piece_at(sq)
|
||||
if pc and pc.piece_type == chess.PAWN and pc.color == color:
|
||||
cnt += 1
|
||||
if cnt > 1:
|
||||
penalty += (cnt - 1) * 12 * (1 if color == chess.WHITE else -1)
|
||||
return penalty
|
||||
|
||||
def _king_safety_bonus(self, king_sq: int, color: chess.Color) -> int:
|
||||
# Bonus for castled-like positions in middlegame; penalty for center-exposed kings
|
||||
if king_sq is None:
|
||||
return 0
|
||||
file_idx = chess.square_file(king_sq)
|
||||
rank_idx = chess.square_rank(king_sq)
|
||||
if color == chess.WHITE:
|
||||
if (file_idx, rank_idx) in {(6, 0), (2, 0)}:
|
||||
return 20
|
||||
if (file_idx, rank_idx) in {(4, 0), (3, 0)}:
|
||||
return -10
|
||||
else:
|
||||
if (file_idx, rank_idx) in {(6, 7), (2, 7)}:
|
||||
return 20
|
||||
if (file_idx, rank_idx) in {(4, 7), (3, 7)}:
|
||||
return -10
|
||||
return 0
|
||||
|
||||
@ -43,6 +43,10 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
|
||||
lf.write(f"game {game_id} started\n")
|
||||
except Exception:
|
||||
game_log_path = None
|
||||
# Simple time manager state
|
||||
my_ms = None
|
||||
opp_ms = None
|
||||
inc_ms = 0
|
||||
try:
|
||||
for event in api.stream_game_events(game_id):
|
||||
et = event.get("type")
|
||||
@ -52,6 +56,10 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
|
||||
state = event.get("state", {})
|
||||
moves = state.get("moves", "")
|
||||
status = state.get("status")
|
||||
# clocks are in milliseconds if present
|
||||
my_ms = state.get("wtime") if color == "white" else state.get("btime")
|
||||
opp_ms = state.get("btime") if color == "white" else state.get("wtime")
|
||||
inc_ms = state.get("winc") or state.get("binc") or 0
|
||||
# Discover my color from gameFull
|
||||
white_id = event["white"].get("id")
|
||||
black_id = event["black"].get("id")
|
||||
@ -64,6 +72,15 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
|
||||
else:
|
||||
moves = event.get("moves", "")
|
||||
status = event.get("status")
|
||||
# update clocks from gameState if present
|
||||
if color == "white":
|
||||
my_ms = event.get("wtime", my_ms)
|
||||
opp_ms = event.get("btime", opp_ms)
|
||||
inc_ms = event.get("winc", inc_ms)
|
||||
elif color == "black":
|
||||
my_ms = event.get("btime", my_ms)
|
||||
opp_ms = event.get("wtime", opp_ms)
|
||||
inc_ms = event.get("binc", inc_ms)
|
||||
|
||||
moves_list = moves.split() if moves else []
|
||||
new_len = len(moves_list)
|
||||
@ -93,12 +110,23 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
|
||||
f"Game {game_id}: turn={'white' if is_white_turn else 'black'}, my_turn={my_turn}"
|
||||
)
|
||||
if my_turn:
|
||||
move, reason = engine.choose_move_with_explanation(board)
|
||||
# Compute a per-move time budget (seconds) based on remaining time
|
||||
# Heuristic: use min( max_time_sec, max(0.05, 0.6 * my_time_left/remaining_moves + inc) )
|
||||
# Estimate remaining moves as 30 - ply/2 bounded to [10, 60]
|
||||
est_moves_left = max(10, min(60, 30 - board.fullmove_number // 2))
|
||||
time_left_sec = (my_ms or 0) / 1000.0
|
||||
inc_sec = (inc_ms or 0) / 1000.0
|
||||
budget = 0.6 * (time_left_sec / max(1, est_moves_left)) + 0.5 * inc_sec
|
||||
# Spend more time per move (requested): double the budget
|
||||
budget *= 2.0
|
||||
# Keep within reasonable bounds
|
||||
budget = max(0.05, min(engine.max_time_sec, budget))
|
||||
move, reason = engine.choose_move_with_explanation(board, time_budget_sec=budget)
|
||||
if move is None:
|
||||
logging.info(f"Game {game_id}: no legal moves (game likely over)")
|
||||
break
|
||||
try:
|
||||
logging.info(f"Game {game_id}: playing {move.uci()}")
|
||||
logging.info(f"Game {game_id}: playing {move.uci()} (budget={budget:.2f}s, my_time_left={time_left_sec:.1f}s, inc={inc_sec:.2f}s)")
|
||||
if game_log_path:
|
||||
with open(game_log_path, "a") as lf:
|
||||
lf.write(f"ply {last_handled_len+1}: {move.uci()}\n{reason}\n\n")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user