kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-07-06 18:23:15 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: log failures in connecting to lichess

Browse Source
This commit is contained in:
Krzysztof kuhy Rudnicki 2025-08-22 21:48:28 +02:00
parent cc0bd572be
commit 59e6f807b7
5 changed files with 152 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
PYTHON/lichess_bot/.bot_version
View File

@ -1 +1 @@
4
10

65
PYTHON/lichess_bot/engine.py
View File

@ -43,6 +43,20 @@ class RandomEngine:
("e2e4", "c7c5"): ["g1f3", "d2d4", "c2c3", "b1c3"],
("d2d4", "d7d5"): ["c2c4", "g1f3", "e2e3"],
("d2d4", "g8f6"): ["c2c4", "g1f3", "e2e3"],
# --- More specific continuations to steer sensible early play ---
# 1.e4 e5 2.Nf3 (Black to move)
("e2e4", "e7e5", "g1f3"): ["b8c6", "g8f6", "f8c5", "d7d6"],
# Italian: 1.e4 e5 2.Nf3 Nc6 3.Bc4 (Black to move)
("e2e4", "e7e5", "g1f3", "b8c6", "f1c4"): ["g8f6", "f8c5", "d7d6"],
# Ruy Lopez: 1.e4 e5 2.Nf3 Nc6 3.Bb5 (Black to move)
("e2e4", "e7e5", "g1f3", "b8c6", "f1b5"): ["a7a6", "g8f6", "f8c5", "d7d6"],
# Scotch: 1.e4 e5 2.Nf3 Nc6 3.d4 (Black to move)
("e2e4", "e7e5", "g1f3", "b8c6", "d2d4"): ["e5d4", "g8f6"],
# Queen's Gambit: 1.d4 d5 2.c4 (Black to move)
("d2d4", "d7d5", "c2c4"): ["e7e6", "c7c6", "d5c4"],
# English: 1.c4 e5 2.Nc3 (Black to move)
("c2c4", "e7e5", "b1c3"): ["g8f6", "b8c6"],
}
def choose_move(self, board: chess.Board, time_budget_sec: Optional[float] = None) -> Optional[chess.Move]:
@ -275,6 +289,30 @@ class RandomEngine:
early = self._is_early_game(board)
piece = board.piece_at(m.from_square)
if piece:
# Discourage premature queen adventures in the opening
if piece.piece_type == chess.QUEEN and early:
victim = board.piece_at(m.to_square)
# Penalize queen pawn-grabs on edge pawns (a2/b2/g2/h2 or a7/b7/g7/h7)
poison_targets_white = {chess.A7, chess.B7, chess.G7, chess.H7}
poison_targets_black = {chess.A2, chess.B2, chess.G2, chess.H2}
is_poison_target = (
(piece.color == chess.WHITE and m.to_square in poison_targets_white)
or (piece.color == chess.BLACK and m.to_square in poison_targets_black)
)
if is_cap and victim and victim.piece_type == chess.PAWN and is_poison_target:
# If destination is heavily attacked, apply a large penalty
attackers_op = len(board.attackers(not piece.color, m.to_square))
defenders_me = len(board.attackers(piece.color, m.to_square))
if attackers_op >= max(1, defenders_me):
s -= 500
else:
s -= 250
# General small penalty for non-check queen moves before minor development
if not is_cap:
if self._most_minors_undeveloped(board, piece.color):
s -= 160
else:
s -= 60
if board.is_castling(m):
s += 650
if piece.piece_type in (chess.KNIGHT, chess.BISHOP):
@ -304,11 +342,20 @@ class RandomEngine:
from_file = chess.square_file(m.from_square)
from_rank = chess.square_rank(m.from_square)
to_rank = chess.square_rank(m.to_square)
# Discourage early f-pawn push and also random wing pawn thrusts like a/b/g/h
if from_file == 5:
if piece.color == chess.WHITE and from_rank == 1 and to_rank == 2:
s -= 140
if piece.color == chess.BLACK and from_rank == 6 and to_rank == 5:
s -= 140
if from_file in (0, 1, 6, 7) and ((piece.color == chess.WHITE and from_rank == 1 and to_rank == 2) or (piece.color == chess.BLACK and from_rank == 6 and to_rank == 5)):
s -= 60
# Discourage early c-pawn push to c4/c5 if we already advanced the e-pawn (prevents e5+c5 blunder-y structures)
if from_file == 2:
e_pawn_sq = chess.E2 if piece.color == chess.WHITE else chess.E7
e_advanced = board.piece_at(e_pawn_sq) is None
if e_advanced and ((piece.color == chess.WHITE and from_rank == 1 and to_rank == 3) or (piece.color == chess.BLACK and from_rank == 6 and to_rank == 4)):
s -= 80
if chess.square_file(m.to_square) in (3, 4):
s += 50
return s
@ -371,6 +418,22 @@ class RandomEngine:
if bk_sq not in (chess.E8, chess.G8, chess.C8):
safety += 40
# Early queen raid penalty: queen deep in opponent camp in the opening
queen_raid_pen = 0
if self._is_early_game(board):
q_w = board.pieces(chess.QUEEN, chess.WHITE)
q_b = board.pieces(chess.QUEEN, chess.BLACK)
if q_w:
qsq = next(iter(q_w))
# White queen on rank 7/8 is often risky early
if chess.square_rank(qsq) >= 6:
queen_raid_pen -= 30
if q_b:
qsq = next(iter(q_b))
# Black queen on rank 1/2 is often risky early
if chess.square_rank(qsq) <= 1:
queen_raid_pen += 30
# Piece-square tendencies (small)
pst = self._pst_score(board)
@ -378,7 +441,7 @@ class RandomEngine:
hanging_pen = self._hanging_pieces_penalty(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 - hanging_pen
white_score = material - dp_pen + mobility_term + center_score + rook_file_bonus + safety + queen_raid_pen + pst - hanging_pen
return white_score if board.turn == chess.WHITE else -white_score
def _opening_book_move(self, board: chess.Board) -> Optional[chess.Move]:

87
PYTHON/lichess_bot/lichess_api.py
View File

@ -20,28 +20,74 @@ class LichessAPI:
"User-Agent": "minimal-lichess-bot/0.1 (+https://lichess.org)"
})
def _request(self, method: str, url: str, *, raise_for_status: bool = False, **kwargs) -> requests.Response:
"""Wrapper around session.request that logs every request/response.
- Logs start (method+URL) and end (status, elapsed).
- On 4xx/5xx, logs a warning with a small snippet of the response body.
- Optionally raises for status.
"""
t0 = time.monotonic()
logging.info(f"HTTP {method} {url} -> sending")
try:
r = self.session.request(method, url, **kwargs)
except Exception as e:
logging.error(f"HTTP {method} {url} -> exception: {e}")
raise
elapsed = time.monotonic() - t0
status = r.status_code
if status >= 400:
# Log a brief error body snippet if available
snippet = None
try:
text = r.text or ""
snippet = text[:200].replace("\n", " ")
except Exception:
snippet = None
if snippet:
logging.warning(f"HTTP {method} {url} -> {status} in {elapsed:.2f}s body='{snippet}'")
else:
logging.warning(f"HTTP {method} {url} -> {status} in {elapsed:.2f}s")
else:
logging.info(f"HTTP {method} {url} -> {status} in {elapsed:.2f}s")
if raise_for_status:
r.raise_for_status()
return r
def stream_events(self) -> Generator[Dict, None, None]:
url = f"{LICHESS_API}/api/stream/event"
with self.session.get(url, stream=True, timeout=60) as r:
r.raise_for_status()
for line in r.iter_lines(decode_unicode=True):
if not line:
backoff = 0.5
while True:
try:
# Use NDJSON Accept and no timeout for long-lived stream
headers = {"Accept": "application/x-ndjson"}
with self._request("GET", url, headers=headers, stream=True, timeout=None) as r:
r.raise_for_status()
backoff = 0.5 # reset on success
for line in r.iter_lines(decode_unicode=True):
if not line:
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
logging.debug(f"Skipping non-JSON line: {line}")
except requests.HTTPError as e:
status = getattr(e.response, "status_code", None)
if status == 429:
logging.warning("Event stream hit 429; backing off")
time.sleep(backoff)
backoff = min(8.0, backoff * 2)
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
logging.debug(f"Skipping non-JSON line: {line}")
raise
def accept_challenge(self, challenge_id: str) -> None:
url = f"{LICHESS_API}/api/challenge/{challenge_id}/accept"
r = self.session.post(url, timeout=30)
r.raise_for_status()
self._request("POST", url, timeout=30, raise_for_status=True)
def decline_challenge(self, challenge_id: str, reason: str = "generic") -> None:
url = f"{LICHESS_API}/api/challenge/{challenge_id}/decline"
data = {"reason": reason}
r = self.session.post(url, data=data, timeout=30)
r.raise_for_status()
self._request("POST", url, data=data, timeout=30, raise_for_status=True)
def join_game_stream(self, game_id: str, my_color: Optional[str]) -> Tuple[chess.Board, str]:
"""Deprecated: use stream_game_events and parse initial state there."""
@ -49,7 +95,8 @@ class LichessAPI:
url = f"{LICHESS_API}/api/board/game/stream/{game_id}"
board = chess.Board()
color = my_color or "white"
with self.session.get(url, stream=True, timeout=60) as r:
headers = {"Accept": "application/x-ndjson"}
with self._request("GET", url, headers=headers, stream=True, timeout=None) as r:
r.raise_for_status()
for line in r.iter_lines(decode_unicode=True):
if not line:
@ -80,7 +127,8 @@ class LichessAPI:
def stream_game_events(self, game_id: str) -> Generator[Dict, None, None]:
url = f"{LICHESS_API}/api/board/game/stream/{game_id}"
with self.session.get(url, stream=True, timeout=60) as r:
headers = {"Accept": "application/x-ndjson"}
with self._request("GET", url, headers=headers, stream=True, timeout=None) as r:
r.raise_for_status()
for line in r.iter_lines(decode_unicode=True):
if not line:
@ -92,10 +140,15 @@ class LichessAPI:
def make_move(self, game_id: str, move: chess.Move) -> None:
url = f"{LICHESS_API}/api/board/game/{game_id}/move/{move.uci()}"
r = self.session.post(url, timeout=30)
r = self._request("POST", url, timeout=30)
if r.status_code in (400, 409):
# Likely not our turn or move already played; do not retry to avoid spam
r.raise_for_status()
return
if r.status_code == 429:
logging.warning(f"HTTP POST {url} -> 429; retrying once after 0.5s")
time.sleep(0.5)
r = self.session.post(url, timeout=30)
r = self._request("POST", url, timeout=30)
r.raise_for_status()
def get_game_state(self, game_id: str) -> Optional[Dict]:
@ -104,7 +157,7 @@ class LichessAPI:
def get_my_user_id(self) -> Optional[str]:
url = f"{LICHESS_API}/api/account"
r = self.session.get(url, timeout=30)
r = self._request("GET", url, timeout=30)
if r.status_code == 200:
return r.json().get("id")
return None

22
PYTHON/lichess_bot/main.py
View File

@ -59,6 +59,9 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
black_name: Optional[str] = None
site_url: Optional[str] = None
try:
# Only send moves on authoritative gameState events to avoid race
# conditions right after gameFull arrives.
seen_game_full = False
for event in api.stream_game_events(game_id):
et = event.get("type")
if et in ("gameFull", "gameState"):
@ -92,6 +95,7 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
elif me == black_id:
color = "black"
logging.info(f"Game {game_id}: joined as {color} (gameFull)")
seen_game_full = True
else:
moves = event.get("moves", "")
status = event.get("status")
@ -132,7 +136,9 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
logging.info(
f"Game {game_id}: turn={'white' if is_white_turn else 'black'}, my_turn={my_turn}"
)
if my_turn:
# Only move on 'gameState' events; skip making a move on initial 'gameFull'
allow_move = (et == "gameState")
if my_turn and allow_move:
# 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]
@ -149,11 +155,15 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
logging.info(f"Game {game_id}: no legal moves (game likely over)")
break
try:
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")
api.make_move(game_id, move)
# Double-check legality just before sending to avoid 400s when state changed.
if move not in board.legal_moves:
logging.info(f"Game {game_id}: selected move no longer legal; skipping send")
else:
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")
api.make_move(game_id, move)
except Exception as e:
logging.warning(f"Game {game_id}: move {move.uci()} failed: {e}")
# Mark this position as handled (whether or not we moved)

2
PYTHON/stockfish_analysis/analyze_chess_game.py
View File

@ -122,7 +122,7 @@ def main():
ap.add_argument("file", help="Path to a PGN file or a log containing a PGN section")
ap.add_argument("--engine", default="stockfish", help="Path to stockfish executable (default: stockfish)")
# Exactly one of time or depth may be provided; default to time
ap.add_argument("--time", type=float, default=0.2, help="Analysis time per evaluation in seconds (default: 0.2)")
ap.add_argument("--time", type=float, default=4, help="Analysis time per evaluation in seconds (default: 0.2)")
ap.add_argument("--depth", type=int, default=None, help="Fixed depth per evaluation (overrides --time)")
args = ap.parse_args()