Add lichess opening explorer

README.md

@@ -87,10 +87,11 @@ Besides the above, there are many possible options within `config.yml` for confi
     - `offer_draw_for_egtb_zero`: If true the bot will offer/accept draw in positions where the online_egtb returns a wdl of 0.
     - `offer_draw_moves`: The absolute value of the evaluation has to be less than or equal to `offer_draw_score` for `offer_draw_moves` amount of moves for the bot to offer/accept draw.
     - `offer_draw_pieces`: The bot only offers/accepts draws if the position has less than or equal to `offer_draw_pieces` pieces.
-- `online_moves`: This section gives your bot access to various online resources for choosing moves like opening books and endgame tablebases. This can be a supplement or a replacement for chess databases stored on your computer. There are three sections that correspond to three different online databases:
+- `online_moves`: This section gives your bot access to various online resources for choosing moves like opening books and endgame tablebases. This can be a supplement or a replacement for chess databases stored on your computer. There are four sections that correspond to four different online databases:
     1. `chessdb_book`: Consults a [Chinese chess position database](https://www.chessdb.cn/), which also hosts a xiangqi database.
     2. `lichess_cloud_analysis`: Consults [Lichess's own position analysis database](https://lichess.org/api#operation/apiCloudEval).
-    3. `online_egtb`: Consults either the online Syzygy 7-piece endgame tablebase [hosted by Lichess](https://lichess.org/blog/W3WeMyQAACQAdfAL/7-piece-syzygy-tablebases-are-complete) or the chessdb listed above.
+    3. `lichess_opening_explorer`: Consults [Lichess's opening explorer](https://lichess.org/api#tag/Opening-Explorer).
+    4. `online_egtb`: Consults either the online Syzygy 7-piece endgame tablebase [hosted by Lichess](https://lichess.org/blog/W3WeMyQAACQAdfAL/7-piece-syzygy-tablebases-are-complete) or the chessdb listed above.
     - `max_out_of_book_moves`: Stop using online opening books after they don't have a move for `max_out_of_book_moves` positions. Doesn't apply to the online endgame tablebases.
     - `max_retries`: The maximum amount of retries when getting an online move.
     - Configurations common to all:
@@ -106,6 +107,11 @@ Besides the above, there are many possible options within `config.yml` for confi
     - Configurations only in `lichess_cloud_analysis`:
         - `max_score_difference`: When `move_quality` is set to `"good"`, this option specifies the maximum difference between the top scoring move and any other move that will make up the set from which a move will be chosen randomly. If this option is set to 25 and the top move in a position has a score of 100, no move with a score of less than 75 will be returned.
         - `min_knodes`: The minimum number of kilonodes to search. The minimum number of nodes to search is this value times 1000.
+    - Configurations only in `lichess_opening_explorer`:
+        - `source`: One of `lichess`, `masters`, or `player`. Whether to use move statistics from masters, lichess players, or a specific player.
+        - `player_name`: Used only when `source` is `player`. The username of the player to use for move statistics.
+        - `sort`: One of `winrate` or `games_played`. Whether to choose the best move according to the winrate or the games played.
+        - `min_games`: The minimum number of times a move must have been played to be considered.
     - Configurations only in `online_egtb`:
         - `max_pieces`: The maximum number of pieces in the current board for which the tablebase will be consulted.
         - `source`: One of `chessdb` or `lichess`. Lichess also has tablebases for atomic and antichess while chessdb only has those for standard.

config.py

@@ -172,6 +172,12 @@ def insert_default_values(CONFIG: CONFIG_DICT_TYPE) -> None:
     set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="min_depth", default=20)
     set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="min_knodes", default=0)
     set_config_default(CONFIG, "engine", "online_moves", "lichess_cloud_analysis", key="max_score_difference", default=50)
+    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="enabled", default=False)
+    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="min_time", default=20)
+    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="source", default="masters")
+    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="player_name", default="")
+    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="sort", default="winrate")
+    set_config_default(CONFIG, "engine", "online_moves", "lichess_opening_explorer", key="min_games", default=10)
     set_config_default(CONFIG, "engine", "lichess_bot_tbs", "syzygy", key="enabled", default=False)
     set_config_default(CONFIG, "engine", "lichess_bot_tbs", "syzygy", key="max_pieces", default=7)
     set_config_default(CONFIG, "engine", "lichess_bot_tbs", "syzygy", key="move_quality", default="best")

config.yml.default

@@ -50,6 +50,13 @@ engine:                      # Engine settings.
       max_score_difference: 50 # Only for move_quality: "good". The maximum score difference (in cp) between the best move and the other moves.
       min_depth: 20
       min_knodes: 0
+    lichess_opening_explorer:
+      enabled: false
+      min_time: 20
+      source: "masters"      # One of "lichess", "masters", "player"
+      player_name: ""        # The lichess username. Leave empty for the bot's username to be used. Used only when source is "player".
+      sort: "winrate"        # One of "winrate", "games_played"
+      min_games: 10          # Minimum number of times a move must have been played to be chosen.
     online_egtb:
       enabled: false
       min_time: 20

engine_wrapper.py

@@ -718,6 +718,7 @@ def get_online_move(li: lichess.Lichess, board: chess.Board, game: model.Game, o
     online_egtb_cfg = online_moves_cfg.online_egtb
     chessdb_cfg = online_moves_cfg.chessdb_book
     lichess_cloud_cfg = online_moves_cfg.lichess_cloud_analysis
+    opening_explorer_cfg = online_moves_cfg.lichess_opening_explorer
     max_out_of_book_moves = online_moves_cfg.max_out_of_book_moves
     offer_draw = False
     resign = False
@@ -742,6 +743,9 @@ def get_online_move(li: lichess.Lichess, board: chess.Board, game: model.Game, o
     if best_move is None and out_of_online_opening_book_moves[game.id] < max_out_of_book_moves:
         best_move, comment = get_lichess_cloud_move(li, board, game, lichess_cloud_cfg)
 
+    if best_move is None and out_of_online_opening_book_moves[game.id] < max_out_of_book_moves:
+        best_move = get_opening_explorer_move(li, board, game, opening_explorer_cfg)
+
     if best_move:
         if isinstance(best_move, str):
             return chess.engine.PlayResult(chess.Move.from_uci(best_move),
@@ -751,7 +755,7 @@ def get_online_move(li: lichess.Lichess, board: chess.Board, game: model.Game, o
                                            resigned=resign)
         return [chess.Move.from_uci(move) for move in best_move]
     out_of_online_opening_book_moves[game.id] += 1
-    used_opening_books = chessdb_cfg.enabled or lichess_cloud_cfg.enabled
+    used_opening_books = chessdb_cfg.enabled or lichess_cloud_cfg.enabled or opening_explorer_cfg.enabled
     if out_of_online_opening_book_moves[game.id] == max_out_of_book_moves and used_opening_books:
         logger.info(f"Will stop using online opening books for game {game.id}.")
     return chess.engine.PlayResult(None, None)
@@ -800,7 +804,7 @@ def get_chessdb_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
 
 def get_lichess_cloud_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
                            lichess_cloud_cfg: config.Configuration) -> tuple[Optional[str], Optional[chess.engine.InfoDict]]:
-    """Get the move from the lichess's cloud analysis."""
+    """Get a move from the lichess's cloud analysis."""
     wb = "w" if board.turn == chess.WHITE else "b"
     time_left = game.state[f"{wb}time"]
     min_time = lichess_cloud_cfg.min_time * 1000
@@ -851,6 +855,51 @@ def get_lichess_cloud_move(li: lichess.Lichess, board: chess.Board, game: model.
     return move, comment
 
 
+def get_opening_explorer_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
+                              opening_explorer_cfg: config.Configuration) -> Optional[str]:
+    """Get a move from lichess's opening explorer."""
+    wb = "w" if board.turn == chess.WHITE else "b"
+    time_left = game.state[f"{wb}time"]
+    min_time = opening_explorer_cfg.min_time * 1000
+    source = opening_explorer_cfg.source
+    if not opening_explorer_cfg.enabled or time_left < min_time or source == "master" and board.uci_variant != "chess":
+        return None
+
+    move = None
+    variant = "standard" if board.uci_variant == "chess" else board.uci_variant
+    try:
+        if source == "masters":
+            params = {"fen": board.fen(), "moves": 100}
+            response = li.online_book_get("https://explorer.lichess.ovh/masters", params)
+        elif source == "player":
+            player = opening_explorer_cfg.player_name
+            if not player:
+                player = game.username
+            params = {"player": player, "fen": board.fen(), "moves": 100, "variant": variant,
+                      "recentGames": 0, "color": "white" if wb == "w" else "black"}
+            response = li.online_book_get("https://explorer.lichess.ovh/player", params, True)
+        else:
+            params = {"fen": board.fen(), "moves": 100, "variant": variant, "topGames": 0, "recentGames": 0}
+            response = li.online_book_get("https://explorer.lichess.ovh/lichess", params)
+        moves = []
+        for possible_move in response["moves"]:
+            games_played = possible_move["white"] + possible_move["black"] + possible_move["draws"]
+            winrate = (possible_move["white"] + possible_move["draws"] * .5) / games_played
+            if games_played >= opening_explorer_cfg.min_games:
+                # We add both winrate and games_played to the tuple, so that if 2 moves are tied on the first metric,
+                # the second one will be used.
+                moves.append((winrate if opening_explorer_cfg.sort == "winrate" else games_played,
+                              games_played if opening_explorer_cfg.sort == "winrate" else winrate, possible_move["uci"]))
+        moves.sort(reverse=True)
+        move = moves[0][2]
+        logger.info(f"Got move {move} from lichess opening explorer ({opening_explorer_cfg.sort}: {moves[0][0]})"
+                    f" for game {game.id}")
+    except Exception:
+        pass
+
+    return move
+
+
 def get_online_egtb_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
                          online_egtb_cfg: config.Configuration) -> tuple[Union[str, list[str], None], int]:
     """

lichess.py

@@ -357,7 +357,7 @@ def cancel(self, challenge_id: str) -> JSON_REPLY_TYPE:
         """Cancel a challenge."""
         return self.api_post("cancel", challenge_id, raise_for_status=False)
 
-    def online_book_get(self, path: str, params: Optional[dict[str, Any]] = None) -> JSON_REPLY_TYPE:
+    def online_book_get(self, path: str, params: Optional[dict[str, Any]] = None, stream: bool = False) -> JSON_REPLY_TYPE:
         """Get an external move from online sources (chessdb or lichess.org)."""
         @backoff.on_exception(backoff.constant,
                               (RemoteDisconnected, ConnectionError, HTTPError, ReadTimeout),
@@ -369,7 +369,7 @@ def online_book_get(self, path: str, params: Optional[dict[str, Any]] = None) ->
                               backoff_log_level=logging.DEBUG,
                               giveup_log_level=logging.DEBUG)
         def online_book_get() -> JSON_REPLY_TYPE:
-            json_response: JSON_REPLY_TYPE = self.other_session.get(path, timeout=2, params=params).json()
+            json_response: JSON_REPLY_TYPE = self.other_session.get(path, timeout=2, params=params, stream=stream).json()
             return json_response
         return online_book_get()