Add chess-style Elo ratings for players

Each player's rating starts at 1500 and updates transactionally with every finished match: a team's rating is the mean of its two members and the standard K=32 formula decides the zero-sum delta applied to both members of a team. Ratings are per game type in a new player_rating table; match_player records each match's elo_delta. - GET /api/leaderboard exposes elo and sorts by it - GET /api/me/matches includes per-player elo deltas - new GET /api/me/ratings returns the caller's rating per game type - frontend: Elo column on the leaderboard, per-match delta in the history page, current rating in the lobby - python -m tavolo.backfill_elo recomputes all ratings from the recorded match history (one-off backfill for existing matches)
2026-09-18 13:36:17 +00:00
parent 3da0c463de
commit a606f14550
15 changed files with 651 additions and 17 deletions
@@ -9,9 +9,10 @@ from httpx import ASGITransport, AsyncClient
 from pwo import async_test

 from tavolo.app import app, tortoise_mixin
+from tavolo.elo import INITIAL_RATING
 from tavolo.game import engine
 from tavolo.game.state import GameState
-from tavolo.models import Match, MatchPlayer
+from tavolo.models import Match, MatchPlayer, PlayerRating
 from tavolo.stats import save_match_result
 from tests.helpers import oidc_user

@@ -48,6 +49,28 @@ def _finished_state() -> GameState:
    return state


+def _finished_state_reversed() -> GameState:
+    """Same one-capture ending as ``_finished_state``, but team B scores it."""
+    state = GameState(
+        id="stats-game-2",
+        join_code="STATS2",
+        creator_sub="alice",
+        target_score=2,
+        phase=engine.PHASE_PLAYING,
+        turn=1,
+        table=[engine.parse_card("02C")],
+    )
+    from tavolo.game.state import PlayerState, Card
+
+    state.players = [
+        PlayerState(sub="alice", name="alice", seat=0),
+        PlayerState(sub="bob", name="bob", seat=1, hand=[Card.parse("02D")]),
+        PlayerState(sub="carol", name="carol", seat=2),
+        PlayerState(sub="dave", name="dave", seat=3),
+    ]
+    return state
+
+
 class SaveMatchResultTest(unittest.TestCase):
    @async_test
    async def test_finished_match_is_persisted_once(self) -> None:
@@ -71,6 +94,58 @@ class SaveMatchResultTest(unittest.TestCase):
            winners = await MatchPlayer.filter(won=True)
            self.assertEqual({"alice", "carol"}, {p.user_sub for p in winners})

+    @async_test
+    async def test_finished_match_updates_elo_ratings(self) -> None:
+        ctx = await _use_app_db()
+        state = _finished_state()
+        engine.play(state, "alice", "02D", ["02C"])
+
+        with ctx:
+            await save_match_result(state)
+
+            ratings = {
+                row.user_sub: row for row in await PlayerRating.all()
+            }
+            self.assertEqual(4, len(ratings))
+            # Four players at 1500: winners gain K/2, losers lose it.
+            for winner in ("alice", "carol"):
+                self.assertEqual(INITIAL_RATING + 16, ratings[winner].rating)
+                self.assertEqual(1, ratings[winner].matches_played)
+            for loser in ("bob", "dave"):
+                self.assertEqual(INITIAL_RATING - 16, ratings[loser].rating)
+                self.assertEqual(1, ratings[loser].matches_played)
+
+            # The per-match delta is recorded on each participation row.
+            deltas = {
+                p.user_sub: p.elo_delta for p in await MatchPlayer.all()
+            }
+            self.assertEqual(
+                {"alice": 16, "carol": 16, "bob": -16, "dave": -16}, deltas
+            )
+
+    @async_test
+    async def test_elo_ratings_accumulate_across_matches(self) -> None:
+        ctx = await _use_app_db()
+        state = _finished_state()
+        engine.play(state, "alice", "02D", ["02C"])
+        reversed_state = _finished_state_reversed()
+        engine.play(reversed_state, "bob", "02D", ["02C"])
+
+        with ctx:
+            await save_match_result(state)
+            # A second match between the same players, won by team B.
+            await save_match_result(reversed_state)
+
+            ratings = {
+                row.user_sub: row.rating for row in await PlayerRating.all()
+            }
+            # Match 1: even teams, team A wins (+16/-16). Match 2: team A
+            # is now the favourite (1516 vs 1484), so losing costs 17.
+            self.assertEqual(INITIAL_RATING - 1, ratings["alice"])
+            self.assertEqual(INITIAL_RATING + 1, ratings["bob"])
+            bob = await PlayerRating.get(user_sub="bob")
+            self.assertEqual(2, bob.matches_played)
+

 async def _seed_two_matches(game_types: tuple = ("scopone_scientifico", "scopone_scientifico")) -> None:
    ctx = await _use_app_db()
@@ -167,6 +242,116 @@ class StatsRouteTest(unittest.TestCase):
        # Alice leads on points after tying Bob on wins.
        self.assertEqual("alice", response.json()["results"][0]["user_sub"])

+    @async_test
+    async def test_leaderboard_includes_elo_and_sorts_by_it(self) -> None:
+        await _seed_two_matches()
+        ctx = await _use_app_db()
+        with ctx:
+            # Bob outranks everyone despite Alice leading on points.
+            await PlayerRating.create(
+                id=uuid.uuid4(),
+                user_sub="bob",
+                game_type="scopone_scientifico",
+                rating=1600,
+                matches_played=2,
+            )
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            response = await client.get("/api/leaderboard")
+        self.assertEqual(200, response.status_code)
+        results = response.json()["results"]
+        by_sub = {row["user_sub"]: row for row in results}
+        self.assertEqual(1600, by_sub["bob"]["elo"])
+        # Players without a rating row report the initial rating.
+        self.assertEqual(INITIAL_RATING, by_sub["alice"]["elo"])
+        # Elo outranks wins/points.
+        self.assertEqual("bob", results[0]["user_sub"])
+
+    @async_test
+    async def test_leaderboard_elo_scoped_by_game_type(self) -> None:
+        await _seed_two_matches()
+        ctx = await _use_app_db()
+        with ctx:
+            await PlayerRating.create(
+                id=uuid.uuid4(),
+                user_sub="alice",
+                game_type="scopone_scientifico",
+                rating=1516,
+                matches_played=1,
+            )
+            # Alice's rating in another game must not leak into the
+            # scopone leaderboard.
+            await PlayerRating.create(
+                id=uuid.uuid4(),
+                user_sub="alice",
+                game_type="other_game",
+                rating=1800,
+                matches_played=1,
+            )
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            response = await client.get("/api/leaderboard?game_type=scopone_scientifico")
+        self.assertEqual(200, response.status_code)
+        results = response.json()["results"]
+        by_sub = {row["user_sub"]: row for row in results}
+        self.assertEqual(1516, by_sub["alice"]["elo"])
+        self.assertEqual(INITIAL_RATING, by_sub["bob"]["elo"])
+
+    @async_test
+    async def test_my_matches_include_elo_delta(self) -> None:
+        ctx = await _use_app_db()
+        state = _finished_state()
+        engine.play(state, "alice", "02D", ["02C"])
+        with ctx:
+            await save_match_result(state)
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            with oidc_user("alice"):
+                response = await client.get("/api/me/matches")
+        self.assertEqual(200, response.status_code)
+        players = {
+            p["user_sub"]: p
+            for p in response.json()["results"][0]["players"]
+        }
+        self.assertEqual(16, players["alice"]["elo_delta"])
+        self.assertEqual(-16, players["bob"]["elo_delta"])
+        self.assertEqual(16, response.json()["results"][0]["your_elo_delta"])
+
+    @async_test
+    async def test_my_ratings_requires_auth(self) -> None:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            response = await client.get("/api/me/ratings")
+        self.assertEqual(401, response.status_code)
+
+    @async_test
+    async def test_my_ratings_returns_only_own_rows(self) -> None:
+        ctx = await _use_app_db()
+        with ctx:
+            await PlayerRating.create(
+                id=uuid.uuid4(),
+                user_sub="alice",
+                game_type="scopone_scientifico",
+                rating=1516,
+                matches_played=1,
+            )
+            await PlayerRating.create(
+                id=uuid.uuid4(),
+                user_sub="bob",
+                game_type="scopone_scientifico",
+                rating=1484,
+                matches_played=1,
+            )
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            with oidc_user("alice"):
+                response = await client.get("/api/me/ratings")
+        self.assertEqual(200, response.status_code)
+        self.assertEqual(
+            [{"game_type": "scopone_scientifico", "rating": 1516, "matches_played": 1}],
+            response.json()["results"],
+        )
+

 class GameTypeFilterTest(unittest.TestCase):
    """Stats endpoints scope results by the match's game type."""