This site runs a chess bot that tries to checkmate you within a set number of moves. To choose its own moves it simulates how a human opponent is likely to reply, using Maia-3, a human move-prediction model from the University of Toronto CSSLab. Maia-3 is licensed under the GNU Affero General Public License v3.0, which asks that people who use the software over a network can get its source. This page is that offer.
Maia-3 runs as its own process. The move-search engine talks to it over a
small text protocol, the same arm's-length way it drives a Stockfish engine
over UCI. The model and the thin wrapper that runs it
(maia_server.py) are the only AGPL-covered part; their complete
source is below. The search engine is a separate program and is published
here too, as a courtesy, in cleaned form.
web app --(not published: private glue)--> move-search engine
| |
UCI pipe ------+ +------ line protocol
v v
Stockfish (GPL-3.0) maia_server.py ---> maia3 (AGPL-3.0, unmodified)
1e13597), AGPL-3.0. Model weights: the
MaiaChess/maia3
collection on Hugging Face.The web application and the layer connecting it to the engine, the opening book and its solver, all game/rating databases, and deployment or operator data. Those are a separate program from the Maia-3 service and/or private data, and are not needed to build or run the covered work.
maia_server.py engine_clients.py sim_race.py ratings.py
The Maia-3 inference subprocess. Runs the unmodified maia3 package behind a process boundary and returns the full move policy. This plus maia3 itself is the AGPL-covered part of the system.
"""Maia-3 inference subprocess -- the isolation boundary.
Maia-3 (https://github.com/CSSLab/maia3) is licensed under the AGPL-3.0.
This program runs the *unmodified* maia3 package in its own process and
exposes a tiny line protocol so that a separate program (the move engine)
can request a human-move probability distribution the same arm's-length way
it drives a UCI engine like Stockfish. Keeping Maia-3 behind this process
boundary is deliberate: the model and this thin wrapper are the only
AGPL-covered part of the system, and their complete source is offered here.
Why not stock `maia3-uci`? The bundled UCI wrapper emits only the top few
candidates as `score cp` info lines and never the underlying policy
probabilities. The simulation engine samples *the whole legal-move
distribution* thousands of times per move, so it needs the full renormalized
policy -- which this wrapper returns.
Protocol (one JSON object per line, request -> response):
-> {"id": 7, "fen": "<FEN>", "self_elo": 1200, "oppo_elo": 2600}
<- {"id": 7, "policy": {"e2e4": 0.21, "g1f3": 0.14, ...}}
-> {"id": 8, "batch": [{"fen": ..., "self_elo": ..., "oppo_elo": ...}, ...]}
<- {"id": 8, "policies": [{...}, {...}, ...]}
`policy` is renormalized over the legal moves of the side to move (UCI in the
real board's coordinates). The batch form exists so a caller running many
concurrent rollouts can amortize GPU forward passes; batching lives entirely
inside this process and does not change the boundary.
The model checkpoint is downloaded once from Hugging Face by the maia3
package (MaiaChess/maia3 collection) into the directory named by the
MAIA3_MODEL_DIR environment variable.
"""
import json
import os
import sys
from collections import deque
import chess
import torch
# Everything below is imported unmodified from the maia3 package.
from maia3.uci import load_model, parse_args
from maia3.dataset import (
get_historical_tokens,
get_legal_moves_mask,
tokenize_board,
)
from maia3.model_registry import ModelResolutionError, resolve_checkpoint_path
from maia3.utils import get_all_possible_moves, mirror_move
MODEL_DIR = os.environ.get("MAIA3_MODEL_DIR", "./maia3_models")
class MaiaModel:
"""Loads a Maia-3 checkpoint once and answers policy queries."""
def __init__(self, model_name=None, device=None):
device = device or ("cuda" if torch.cuda.is_available() else "cpu")
# 79M on GPU, 23M on CPU: on GPU the forward pass is overhead-bound,
# so the larger model is effectively free; on CPU it is compute-bound.
model_name = model_name or ("maia3-79m" if device == "cuda"
else "maia3-23m")
cfg = parse_args(["--model", model_name, "--device", device])
try:
cfg.checkpoint_path = resolve_checkpoint_path(
cfg.model_spec, cache_dir=MODEL_DIR, local_files_only=True)
except ModelResolutionError:
cfg.checkpoint_path = resolve_checkpoint_path(
cfg.model_spec, cache_dir=MODEL_DIR)
self.cfg = cfg
self.model = load_model(cfg)
self.model.eval()
self._moves = get_all_possible_moves()
self._move_to_idx = {m: i for i, m in enumerate(self._moves)}
def _tokens(self, board):
hist = deque(maxlen=self.cfg.history)
hist.append(tokenize_board(board))
return get_historical_tokens(
hist, self.cfg, base=0.0, inc=0.0,
clk_left_before=0.0, clk_ponder=0.0)
@torch.no_grad()
def policies(self, requests):
"""requests: list of (board, self_elo, oppo_elo). Returns a list of
{uci: prob} dicts renormalized over each board's legal moves. One
batched forward pass."""
if not requests:
return []
dev = self.cfg.device
toks = torch.stack([self._tokens(b) for b, _, _ in requests]).to(dev)
se = torch.tensor([s for _, s, _ in requests],
dtype=torch.long, device=dev)
oe = torch.tensor([o for _, _, o in requests],
dtype=torch.long, device=dev)
logits_move, _, _ = self.model(toks, se, oe)
logits_move = logits_move.float().cpu()
out = []
for row, (board, _, _) in zip(logits_move, requests):
probs = torch.softmax(row, dim=-1)
dist = {}
black = board.turn == chess.BLACK
for mv in board.legal_moves:
uci = mv.uci()
# The model space is from the side-to-move's view (mirrored
# for Black), so look the move up in mirrored coordinates.
key = mirror_move(uci) if black else uci
idx = self._move_to_idx.get(key)
dist[uci] = float(probs[idx]) if idx is not None else 0.0
total = sum(dist.values())
if total > 0:
dist = {k: v / total for k, v in dist.items()}
else: # numerically degenerate: fall back to uniform
n = len(dist) or 1
dist = {k: 1.0 / n for k in dist}
out.append(dist)
return out
def _board(fen):
return chess.Board(fen) if fen else chess.Board()
def main():
model = MaiaModel(os.environ.get("MAIA3_MODEL"))
print("maia-ready", file=sys.stderr, flush=True)
for raw in sys.stdin:
line = raw.strip()
if not line:
continue
try:
req = json.loads(line)
except ValueError:
continue
if req.get("cmd") == "quit":
return
rid = req.get("id")
try:
if "batch" in req:
items = [(_board(x["fen"]),
int(x.get("self_elo", 1500)),
int(x.get("oppo_elo", 1500))) for x in req["batch"]]
resp = {"id": rid, "policies": model.policies(items)}
else:
item = (_board(req["fen"]),
int(req.get("self_elo", 1500)),
int(req.get("oppo_elo", 1500)))
resp = {"id": rid, "policy": model.policies([item])[0]}
except Exception as exc: # never wedge the caller on one bad request
resp = {"id": rid, "error": str(exc)}
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
if __name__ == "__main__":
main()
Arm's-length clients: the search drives Maia-3 (subprocess) and Stockfish (UCI) as separate child processes. The SF11 Contempt=100 anchor is an optional toggle here.
"""Arm's-length clients for the two engines the move-search drives.
Both are separate programs the search talks to over a pipe:
* MaiaModelClient -> the Maia-3 inference subprocess (maia_server.py),
returning a human-move probability distribution.
* StockfishClient -> a Stockfish binary over UCI, for exact evaluation
and the "anchor" ranking.
Neither engine is linked into this process; they run as child processes and
are spoken to over stdin/stdout, exactly as a chess GUI drives a UCI engine.
"""
import json
import os
import subprocess
import sys
import chess
import chess.engine
class MaiaModelClient:
"""Speaks the maia_server.py line protocol to a Maia-3 subprocess."""
def __init__(self, python=None, server=None, env=None):
python = python or sys.executable
server = server or os.path.join(os.path.dirname(__file__),
"maia_server.py")
self._proc = subprocess.Popen(
[python, server],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
text=True, bufsize=1, env=env or os.environ.copy())
self._id = 0
# Block until the model reports ready (printed to the child's stderr,
# which is inherited, so it simply surfaces in the console).
def policy(self, board, self_elo, oppo_elo):
"""{uci: prob} over `board`'s legal moves for the side to move."""
return self._batch([(board, self_elo, oppo_elo)])[0]
def policies(self, requests):
"""Batched form: requests is a list of (board, self_elo, oppo_elo).
Sending concurrent rollout queries together lets the server amortize
the GPU forward pass -- the single biggest throughput lever."""
return self._batch(requests)
def _batch(self, requests):
self._id += 1
payload = {"id": self._id, "batch": [
{"fen": b.fen(), "self_elo": int(s), "oppo_elo": int(o)}
for b, s, o in requests]}
self._proc.stdin.write(json.dumps(payload) + "\n")
self._proc.stdin.flush()
resp = json.loads(self._proc.stdout.readline())
if "error" in resp:
raise RuntimeError("maia subprocess: " + resp["error"])
return resp["policies"]
def close(self):
try:
self._proc.stdin.write('{"cmd": "quit"}\n')
self._proc.stdin.flush()
except Exception:
self._proc.kill()
# ---------------------------------------------------------------------------
# Stockfish, with the optional SF11-contempt "anchor".
#
# The search uses Stockfish two ways:
# 1. Exact evaluation / candidate ranking during rollouts and at the root.
# 2. A "cruise" anchor: while still far from the mate deadline, play a
# solid move that keeps winning chances alive instead of simplifying
# into a drawish endgame.
#
# For (2) an older Stockfish (v11, the last with a real `Contempt` option)
# set to Contempt=100 is a cheap, strong way to make the bot avoid drawish
# liquidations. It is optional and controlled by USE_CONTEMPT_ANCHOR below.
#
# Turn the contempt anchor ON when:
# * the machine running the engine is underpowered -- a shallow contempt
# search is far cheaper than a full Maia-3 simulation race and still
# steers the game well; and/or
# * par is very long (roughly > 60 moves). At long horizons the Maia-3
# rollouts rarely reach a mate before the deadline, so mate-rate ranking
# becomes noise (see sim_race.py) and the contempt anchor is the more
# reliable signal.
# Turn it OFF (use only the modern engine at Contempt=0) when you have the
# compute for full simulation races at ordinary par lengths.
# ---------------------------------------------------------------------------
USE_CONTEMPT_ANCHOR = os.environ.get("USE_CONTEMPT_ANCHOR", "1") != "0"
# Point these at your binaries via the environment; no paths are hard-coded.
STOCKFISH_PATH = os.environ.get("STOCKFISH_PATH", "stockfish")
STOCKFISH_CONTEMPT_PATH = os.environ.get("STOCKFISH_CONTEMPT_PATH", "") # e.g. an SF11 build
class StockfishClient:
"""Thin wrapper over a UCI engine binary."""
def __init__(self, path=None, threads=1, hash_mb=256, options=None):
self.engine = chess.engine.SimpleEngine.popen_uci(
path or STOCKFISH_PATH)
cfg = {"Threads": threads, "Hash": hash_mb}
if options:
cfg.update(options)
try:
self.engine.configure(cfg)
except chess.engine.EngineError:
# Older engines reject unknown options; keep the ones they accept.
for k, v in cfg.items():
try:
self.engine.configure({k: v})
except chess.engine.EngineError:
pass
def best(self, board, depth=12):
return self.engine.play(
board, chess.engine.Limit(depth=depth)).move
def multipv(self, board, count, depth=10):
infos = self.engine.analyse(
board, chess.engine.Limit(depth=depth), multipv=count)
return infos if isinstance(infos, list) else [infos]
def close(self):
try:
self.engine.quit()
except Exception:
pass
def make_contempt_anchor():
"""The optional cruise/ranking anchor. Returns a StockfishClient at
Contempt=100 when enabled and a contempt-capable binary is configured,
else None (callers then fall back to the modern engine)."""
if not USE_CONTEMPT_ANCHOR or not STOCKFISH_CONTEMPT_PATH:
return None
return StockfishClient(STOCKFISH_CONTEMPT_PATH,
options={"Contempt": 100})
The simulation-race move search: for each candidate first move, simulate human-like continuations with Maia-3 and keep the move with the best mate-by-deadline rate. Cleaned reference.
"""Simulation-race attacker (cleaned reference).
The bot plays White and must force checkmate on or before a fixed move
deadline ("par"). Against a human-like defender there is usually no single
forced mate, so instead of a minimax search the engine runs a Monte-Carlo
*race*: for each candidate first move it simulates many plausible
continuations and measures how often that move leads to mate by the
deadline, then keeps the candidate with the best mate rate.
The defender in each simulation is Maia-3 (a human move-prediction model),
so "mate rate" estimates how often a real human of the target rating would
get mated -- not whether mate is forced. This is what makes the bot's play
feel like it is hunting a human king rather than solving a study.
Pipeline per move:
book? -> if the position is in the opening book, play the book move.
cruise -> if still far from par, play a solid anchor move (optionally the
Stockfish contempt anchor) and save the simulation budget.
race -> otherwise run the simulation race below.
This module is a self-contained, readable reference for the race itself. It
talks to Maia-3 and Stockfish only through the arm's-length clients in
engine_clients.py.
"""
import math
import random
import chess
from engine_clients import MaiaModelClient, StockfishClient, make_contempt_anchor
from ratings import maia_elo_for, par_moves
# --- How many simulations are "enough"? ------------------------------------
#
# A candidate's mate rate is a Binomial estimate: run N rollouts, m end in
# mate by par, rate = m / N with standard error sqrt(p(1-p)/N). The race
# only needs to *rank* candidates, so it runs until the leader is separated
# from the runner-up by more than a few standard errors -- not until every
# rate is pinned down.
#
# Practical guidance (watch these while tuning):
#
# * MIN_SIMS ~ 200 per candidate before a mate rate means anything. Below
# ~30 rollouts the estimate is mostly noise; do not rank on it.
#
# * Separation, not precision. Stop a pairwise comparison once
# |p_lead - p_next| > Z * sqrt( p(1-p)/N ) (Z ~= 2.3 for ~99%)
# Two candidates whose intervals still overlap are a genuine tie -- break
# ties by the exact (Stockfish) evaluation, not by more rollouts.
#
# * SATURATION is the failure mode to watch for. If nearly every candidate
# mates a large fraction of the time (say > 0.75), the horizon is too
# generous for the defender's strength: every move looks winning and the
# ranking is noise. This shows up at low target ratings with long par.
# When it happens, trust the anchor evaluation and/or shorten the effective
# horizon rather than believing a 0.86-vs-0.84 mate-rate "win".
#
# * LONG PAR (roughly > 60 moves) is the opposite failure: rollouts almost
# never reach mate before the deadline, so mate rates collapse toward 0
# with high variance and carry little signal. Lean on the contempt anchor
# there (see engine_clients.USE_CONTEMPT_ANCHOR).
#
# * A whole-move budget of a few hundred to a few thousand rollouts total
# is typical; the hard cap is roughly (#candidates * tie-resolution cost).
# Time, not a fixed sim count, is usually the real limit in live play.
MIN_SIMS = 200 # floor per candidate before ranking on mate rate
ROUND_SIMS = 100 # rollouts added per surviving candidate per round
Z = 2.33 # ~99% one-sided separation threshold
TIE_MARGIN = 0.015 # mate-rate gaps below this are treated as ties
SATURATION = 0.75 # above this, suspect the horizon is too generous
ROOT_MULTIPV = 24 # candidate first moves to consider (from Stockfish)
W70_CP = 70 # White rollout moves: sample among moves within 70cp
CRUISE_MOVES_LEFT = 12 # cruise (don't race) while more than this to par
def _std_err(p, n):
return math.sqrt(max(p * (1.0 - p), 1e-9) / max(n, 1))
class SimRaceAttacker:
def __init__(self, deadline, rating, time_control="rapid", rng_seed=None):
self.deadline = int(deadline) # absolute mate-by full-move
self.self_elo = maia_elo_for(rating, time_control) # defender model
self.oppo_elo = 2600 # defender "thinks" it faces a GM
self.maia = MaiaModelClient()
self.sf = StockfishClient() # modern engine, exact eval
self.anchor = make_contempt_anchor() # optional SF11 Contempt=100
self.rng = random.Random(rng_seed)
def moves_left(self, board):
return max(0, self.deadline - board.fullmove_number + 1)
# -- candidate first moves ---------------------------------------------
def _candidates(self, board):
"""The plausible first moves to race. Take Stockfish's multi-PV and
drop only the worst; a move that looks bad to an engine can still be
the fastest maters against a human (so we do not prune hard)."""
infos = self.sf.multipv(board, ROOT_MULTIPV, depth=10)
moves = [inf["pv"][0] for inf in infos if inf.get("pv")]
# De-duplicate positions that transpose to somewhere already seen in
# the game would go here (anti-perpetual); omitted from this excerpt.
return moves or list(board.legal_moves)
# -- one rollout to the deadline ---------------------------------------
def _rollout(self, board, first_move):
"""Play `first_move`, then: Black = Maia-3 sample, White = a sample
among near-best Stockfish moves, until mate / game end / deadline.
Returns True iff White delivers mate on or before par."""
b = board.copy()
b.push(first_move)
while not b.is_game_over():
if b.fullmove_number > self.deadline and b.turn == chess.WHITE:
return False # deadline passed, no mate
if b.turn == chess.WHITE:
b.push(self._white_move(b))
else:
b.push(self._black_move(b))
return b.is_checkmate() and b.turn == chess.BLACK # White mated Black
def _white_move(self, board):
# Uniform sample among Stockfish moves within W70_CP of best: keeps
# White strong but not deterministic, so rollouts explore.
infos = self.sf.multipv(board, 5, depth=8)
best = infos[0]["score"].pov(chess.WHITE).score(mate_score=100000)
window = [inf["pv"][0] for inf in infos
if inf.get("pv") and
best - inf["score"].pov(chess.WHITE).score(mate_score=100000)
<= W70_CP]
return self.rng.choice(window or [infos[0]["pv"][0]])
def _black_move(self, board):
# The human-like defender: sample from Maia-3's full policy.
dist = self.maia.policy(board, self.self_elo, self.oppo_elo)
moves, weights = zip(*dist.items())
return chess.Move.from_uci(self.rng.choices(moves, weights)[0])
# -- the race ----------------------------------------------------------
def choose(self, board, max_rounds=40):
"""Pick the move with the best mate-by-par rate. Runs rounds of
rollouts, eliminating clearly-beaten candidates, until one remains or
the field is a statistical tie (then the exact eval breaks it)."""
if self.moves_left(board) > CRUISE_MOVES_LEFT:
return self._cruise(board) # too far out to race
stats = {m: [0, 0] for m in self._candidates(board)} # move -> [sims, mates]
for _ in range(max_rounds):
for mv in list(stats):
for _ in range(ROUND_SIMS if stats[mv][0] else MIN_SIMS):
stats[mv][0] += 1
stats[mv][1] += 1 if self._rollout(board, mv) else 0
if self._saturated(stats):
break # ranking is noise; stop early
survivors = self._eliminate(stats)
if len(survivors) <= 1:
break
stats = {m: stats[m] for m in survivors}
return self._winner(board, stats)
def _rate(self, s):
return s[1] / s[0] if s[0] else 0.0
def _saturated(self, stats):
return all(self._rate(s) > SATURATION for s in stats.values())
def _eliminate(self, stats):
"""Keep candidates not clearly beaten by the current leader."""
lead = max(stats, key=lambda m: self._rate(stats[m]))
lp, ln = self._rate(stats[lead]), stats[lead][0]
keep = []
for m, s in stats.items():
p, n = self._rate(s), s[0]
gap = lp - p
se = _std_err((lp + p) / 2, min(ln, n))
if m == lead or gap < TIE_MARGIN or gap < Z * se:
keep.append(m) # still in contention
return keep
def _winner(self, board, stats):
"""Best mate rate; ties (within TIE_MARGIN) broken by exact eval."""
best = max(self._rate(s) for s in stats.values())
tied = [m for m, s in stats.items()
if best - self._rate(s) <= TIE_MARGIN]
if len(tied) == 1:
return tied[0]
# Tie -> soundest by the anchor if present, else the modern engine.
eng = self.anchor or self.sf
return max(tied, key=lambda m: self._eval_after(eng, board, m))
def _eval_after(self, eng, board, move):
b = board.copy()
b.push(move)
infos = eng.multipv(b, 1, depth=10)
return -infos[0]["score"].pov(chess.WHITE).score(mate_score=100000)
# -- cruise ------------------------------------------------------------
def _cruise(self, board):
"""Far from par: play a solid move and save the simulation budget.
Prefer the contempt anchor (avoids drawish liquidations) when it is
enabled; otherwise the modern engine."""
eng = self.anchor or self.sf
return eng.best(board, depth=12)
def close(self):
self.maia.close()
self.sf.close()
if self.anchor:
self.anchor.close()
def par_for(rating, time_control="rapid"):
"""Convenience: the mate-by deadline for an entered rating."""
return par_moves(rating)
Rating conversions and the mate-by-par curve. Same par for the same number in both time controls; time control only selects the Maia conditioning table.
"""Rating conversions and the mate-by-par curve.
Players enter a chess.com rating (rapid or blitz). Internally three scales
are used:
* chess.com rapid/blitz -- what the player types on the slider.
* Lichess blitz -- what the Maia-3 model is conditioned on
(Maia-3 is trained on Lichess *blitz* games).
* "par" -- the mate-by move deadline shown as the goal.
Two design rules live here:
1. Same number, same par. A rating of 800 gets the same par whether the
player picked rapid or blitz -- the time control does not change the
goal. The only place time control matters is which anchor table maps
the entered rating to a Lichess-blitz conditioning value for Maia.
2. Everything is a straight piecewise-linear interpolation over published
anchor points, so the curves never silently drift apart between the
engine and the UI (the UI mirrors cc_par exactly).
The anchor points are derived from public rating-comparison tables; swap in
your own if you have better data for your population.
"""
import math
# --- Mate-by-par -----------------------------------------------------------
# Linear from (400 -> 15 moves) to (2400 -> 50 moves): the stronger the
# defender, the more moves you are given to force mate. 2400 is the ceiling
# (the model's useful conditioning tops out around there).
PAR_LO = (400, 15)
PAR_HI = (2400, 50)
PAR_SLOPE = (PAR_HI[1] - PAR_LO[1]) / (PAR_HI[0] - PAR_LO[0]) # 0.0175
def par_moves(rating, lo=400, hi=2400):
"""Mate-by deadline (full moves) for a given entered rating. Clamped to
[lo, hi]; round-half-up. Identical for rapid and blitz."""
r = max(lo, min(hi, int(rating)))
return int(math.floor(PAR_LO[1] + PAR_SLOPE * (r - PAR_LO[0]) + 0.5))
# --- chess.com rapid -> Lichess blitz (Maia-3 conditioning) ----------------
# Used when the player entered a RAPID rating.
_CC_RAPID_TO_LI_BLITZ = [
(400, 880), (735, 1030), (835, 1075), (945, 1145), (1035, 1200),
(1130, 1335), (1230, 1420), (1320, 1475), (1405, 1565), (1500, 1635),
(1575, 1705), (1655, 1780), (1730, 1850), (1810, 1910), (1890, 1970),
(1990, 2050), (2035, 2100), (2080, 2170), (2135, 2235), (2190, 2295),
(2235, 2370), (2290, 2445), (2355, 2560), (2425, 2625), (2480, 2695),
]
# chess.com blitz -> Lichess blitz (Maia-3 conditioning).
# Used when the player entered a BLITZ rating. This is the composition of a
# blitz->rapid table with the rapid->li-blitz table above, so a blitz rating
# maps to the same Lichess-blitz strength it would have via the two-step
# chain -- the model each mode faces is unchanged; only the entry scale
# differs.
_CC_BLITZ_TO_LI_BLITZ = [
(400, 985), (500, 1030), (600, 1075), (700, 1145), (800, 1200),
(900, 1335), (1000, 1420), (1100, 1475), (1200, 1565), (1300, 1635),
(1400, 1705), (1500, 1780), (1600, 1850), (1700, 1910), (1800, 1970),
(1900, 2050), (2000, 2100), (2100, 2170), (2200, 2235), (2300, 2295),
(2400, 2370), (2500, 2445), (2600, 2560), (2700, 2625), (2800, 2695),
]
# Maia-3's trained conditioning saturates around here; past it the elo
# embedding extrapolates silently, so cap what is fed to the model.
MAIA_ELO_MAX = 2600
def _interp(table, x):
x = max(table[0][0], min(table[-1][0], int(x)))
for (x0, y0), (x1, y1) in zip(table, table[1:]):
if x <= x1:
return round(y0 + (y1 - y0) * (x - x0) / (x1 - x0))
return table[-1][1]
def maia_elo_for(rating, time_control="rapid"):
"""Entered chess.com rating -> Lichess-blitz conditioning for Maia-3,
capped at MAIA_ELO_MAX. `time_control` selects the anchor table; it is
the ONLY place the pipeline is time-control aware."""
table = _CC_BLITZ_TO_LI_BLITZ if time_control == "blitz" \
else _CC_RAPID_TO_LI_BLITZ
return min(MAIA_ELO_MAX, _interp(table, rating))