Рейтинг: движок Elo с множителем отрыва и тесты на примеры документа

scoring.py получает движок из docs/rating/rating-system.md: ожидание пары,
K 64 → 16 за 20 партий, вес стола G(N), множитель отрыва (темп, цели, миры,
clamp [0.5, 2]) и близость по типу победы, включая last_standing. rate_match
и replay — чистые функции без БД; replay отдаёт рейтинги без округления,
ΔR и результат относительно ожидания по каждой партии.

Тесты: 15 примеров раздела 6 с числами документа, совпадение констант
и всех ΔR сезона с эталоном simulate.py (полные партии и история без деталей),
монотонность и сумма-ноль. League Points пока остаётся в модуле — витрины
переводятся отдельным коммитом. #23

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LqSoRj99iwVEH5U5fnZgsd
This commit is contained in:
2026-09-14 22:19:40 +03:00
co-authored by Claude Opus 5
parent e40d93e7be
commit e58b4f6614
2 changed files with 395 additions and 21 deletions
+198 -21
View File
@@ -1,35 +1,217 @@
"""Метрика рейтинга. Вынесена отдельно — легко заменить.
"""Метрика рейтинга: многопользовательский Elo с множителем отрыва (#22, #23).
По умолчанию: League Points — нормированные очки за место с учётом размера стола
и ничьих (competition ranking). За партию из N игроков:
points = (N - place - (tie_size - 1)/2) / (N - 1)
1-е место = 1.0, последнее = 0.0; равные места делят сумму очков поровну.
Полное описание, обоснование коэффициентов и примеры — docs/rating/rating-system.md;
эталонная реализация тех же формул — docs/rating/simulate.py (тесты сверяют с ней).
Рейтинговый счёт игрока — сглаженное среднее (байесовское, формула IMDB):
score = (PRIOR_GAMES * PRIOR_MEAN + SUM(points)) / (PRIOR_GAMES + games) * 100
К реальным партиям «дописываются» PRIOR_GAMES виртуальных со средним PRIOR_MEAN:
на малой выборке рейтинг держится около 50 и лишь с опытом сходится к чистому
среднему — короткая удачная серия новичка не обгоняет стабильного ветерана.
Партия раскладывается на пары игроков. Для пары a (выше или наравне) и b:
E_ab = 1 / (1 + 10^((R_b − R_a) / D)) ожидание по рейтингам ДО партии
S_ab = 1 / 0.5 / 0 выше / поровну / ниже
ΔR_i = K_i · G(N) / (N − 1) · Σ_j M_ij · (S_ij − E_ij)
K_i спускается от K_MAX у новичка до K_MIN за K_GAMES партий, G(N) — вес размера стола,
M — множитель отрыва (темп, цели, миры; близость по типу победы — только у пар
с победителем). Недостающий признак партии подставляется типичным и не влияет на M.
Модуль — только константы и чистые функции без БД: калибровка на реальных данных —
правка констант, пересчёт выполняется сам (рейтинг — функция упорядоченной истории).
"""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import date, datetime
from itertools import combinations
# Порог числа игр для попадания в ранжированный топ (ниже — «Новички»/provisional).
MIN_GAMES = 10
# Порог числа игр на фракцию для расчёта лучшей/худшей фракции.
FACTION_MIN_GAMES = 2
# Сглаживание рейтинга: сколько «виртуальных» партий и с каким средним добавляем.
# ─── Правила игры ────────────────────────────────────────────────────────────
# Размер поля в тайлах по числу игроков (дуэль — 2×3, шестеро — 4×5).
BOARD_TILES = {2: 6, 3: 9, 4: 12, 5: 16, 6: 20}
WORLDS_PER_TILE = 2.2
BASE_ROUNDS = 8
# Домашнее правило группы: при 5–6 игроках играется 9 раундов.
EXTENDED_ROUNDS = 9
EXTENDED_MIN_PLAYERS = 5
# ─── Коэффициенты (документ, 4.10) ───────────────────────────────────────────
R0 = 1500.0 # стартовый рейтинг
D = 400.0 # масштаб: разница 400 пунктов — шансы 10:1
K_MAX = 64.0 # K новичка (0 партий)
K_MIN = 16.0 # K опытного игрока
K_GAMES = 20 # за сколько партий K линейно спускается от K_MAX к K_MIN
W_TABLE = 0.5 # вес размера стола
W_TEMPO = 1.0 # вес темпа победы
W_OBJ = 0.5 # вес отрыва по целям
W_WORLDS = 0.5 # вес отрыва по мирам
MU_OBJ = 0.5 # типичный отрыв по целям
MU_WORLDS = 0.5 # типичный отрыв по мирам
M_MIN = 0.5 # страховка: одна партия не легче половины обычной…
M_MAX = 2.0 # …и не тяжелее двух
# Близость партии по типу победы — множитель пар с победителем.
CLOSENESS = {
"objectives": 1.0,
"worlds": 0.85,
"plastic": 0.7,
"resources": 0.6,
"last_standing": 1.0,
}
def max_rounds(player_count: int, nine_rounds_rule: bool) -> int:
"""Лимит раундов партии: 9 при хоумруле группы и 5+ игроках, иначе 8."""
if nine_rounds_rule and player_count >= EXTENDED_MIN_PLAYERS:
return EXTENDED_ROUNDS
return BASE_ROUNDS
def fair_worlds(player_count: int) -> float:
"""«Честная доля» миров на игрока — масштаб для разницы миров."""
return BOARD_TILES[player_count] * WORLDS_PER_TILE / player_count
def mu_tempo(rmax: int) -> float:
"""Типичный темп: партия закончилась в предпоследнем раунде."""
return 1.0 / (rmax - 1)
def expected(r_a: float, r_b: float) -> float:
"""Ожидаемый результат a против b (вероятность, что a окажется выше)."""
return 1.0 / (1.0 + 10.0 ** ((r_b - r_a) / D))
def k_factor(games: int) -> float:
left = max(0.0, 1.0 - games / K_GAMES)
return K_MIN + (K_MAX - K_MIN) * left
def table_weight(player_count: int) -> float:
return 1.0 + W_TABLE * (player_count - 2) / 4.0
def _clamp(x: float, lo: float, hi: float) -> float:
return max(lo, min(hi, x))
# ─── Партия как вход расчёта ─────────────────────────────────────────────────
@dataclass(frozen=True)
class RatedSeat:
user_id: int
place: int
faction_id: int = 0
eliminated: bool = False
objectives: int | None = None # маркеры целей на конец партии
worlds: int | None = None # дружественные миры на конец партии
@dataclass(frozen=True)
class RatedMatch:
seats: tuple[RatedSeat, ...]
win_reason: str | None = None
end_round: int | None = None # раунд, в котором партия закончилась
nine_rounds_rule: bool = False # снимок настройки группы на момент партии
id: int = 0
group_id: int = 0
played_at: date | None = None
finished_at: datetime | None = None
def pair_multiplier(m: RatedMatch, a: RatedSeat, b: RatedSeat) -> float:
"""Множитель отрыва пары; a — выше или наравне с b."""
n = len(m.seats)
tie = a.place == b.place
winner_pair = a.place == 1
add = 1.0
def diff(x: int, y: int) -> float:
return abs(x - y) if tie else max(0, x - y)
if winner_pair:
rmax = max_rounds(n, m.nine_rounds_rule)
mu = mu_tempo(rmax)
tempo = mu if m.end_round is None else (rmax - m.end_round) / (rmax - 1)
add += W_TEMPO * (tempo - mu)
if winner_pair and m.win_reason == "last_standing":
obj = 1.0 # все соперники устранены — отрыв максимальный, сколько бы ни было маркеров
elif a.objectives is not None and b.objectives is not None:
obj = _clamp(diff(a.objectives, b.objectives) / n, 0.0, 1.0)
else:
obj = MU_OBJ
add += W_OBJ * (obj - MU_OBJ)
if a.worlds is not None and b.worlds is not None:
wor = _clamp(diff(a.worlds, b.worlds) / fair_worlds(n), 0.0, 1.0)
else:
wor = MU_WORLDS
add += W_WORLDS * (wor - MU_WORLDS)
close = CLOSENESS.get(m.win_reason, 1.0) if winner_pair and m.win_reason else 1.0
return _clamp(add, M_MIN, M_MAX) * close
def rate_match(
ratings: dict[int, float], games: dict[int, int], m: RatedMatch
) -> tuple[dict[int, float], dict[int, float]]:
"""Изменения рейтинга участников и их результат относительно ожидания.
Возвращает (ΔR, perf): perf_i = Σ_j (S_ij − E_ij) / (N − 1) — насколько игрок
выступил выше ожидания, без множителя отрыва и K. Входные словари не мутирует."""
n = len(m.seats)
delta = {s.user_id: 0.0 for s in m.seats}
perf = {s.user_id: 0.0 for s in m.seats}
if n < 2:
return delta, perf
g = table_weight(n)
r = {s.user_id: ratings.get(s.user_id, R0) for s in m.seats}
k = {s.user_id: k_factor(games.get(s.user_id, 0)) for s in m.seats}
for a, b in combinations(m.seats, 2):
if a.place > b.place:
a, b = b, a
s_ab = 0.5 if a.place == b.place else 1.0
e_ab = expected(r[a.user_id], r[b.user_id])
x = pair_multiplier(m, a, b) * (s_ab - e_ab)
delta[a.user_id] += k[a.user_id] * g / (n - 1) * x
delta[b.user_id] -= k[b.user_id] * g / (n - 1) * x
perf[a.user_id] += (s_ab - e_ab) / (n - 1)
perf[b.user_id] -= (s_ab - e_ab) / (n - 1)
return delta, perf
@dataclass
class Replay:
"""Итог проигрывания истории: рейтинги без округления и следы каждой партии."""
ratings: dict[int, float] = field(default_factory=dict)
games: dict[int, int] = field(default_factory=dict)
delta: dict[tuple[int, int], float] = field(default_factory=dict) # (match_id, user_id)
perf: dict[tuple[int, int], float] = field(default_factory=dict) # (match_id, user_id)
def replay(matches: Iterable[RatedMatch]) -> Replay:
"""Проигрывает партии в переданном порядке (хронологию задаёт вызывающий)."""
out = Replay()
for m in matches:
delta, perf = rate_match(out.ratings, out.games, m)
for uid, dv in delta.items():
out.ratings[uid] = out.ratings.get(uid, R0) + dv
out.games[uid] = out.games.get(uid, 0) + 1
out.delta[(m.id, uid)] = dv
out.perf[(m.id, uid)] = perf[uid]
return out
# ─── League Points (прежняя система; уходит вместе с SQL-витринами) ──────────
PRIOR_GAMES = 10
PRIOR_MEAN = 0.5
# SQL-выражение сглаженного рейтинга поверх агрегата по строкам scored (s.points).
# При 0 партий SUM = NULL → score = NULL (рейтинга без игр нет).
SMOOTHED_SCORE_SQL = (
f"({PRIOR_GAMES} * {PRIOR_MEAN} + SUM(s.points)) / ({PRIOR_GAMES} + COUNT(*)) * 100"
)
# SQL-выражение очков за участие (tie-aware). Использует поля m.player_count,
# mp.place и t.tie_size (размер группы игроков с тем же местом в партии).
MATCH_POINTS_SQL = (
"CASE WHEN m.player_count > 1 "
"THEN (m.player_count - mp.place - (t.tie_size - 1) / 2.0) "
@@ -39,13 +221,8 @@ MATCH_POINTS_SQL = (
def smoothed_score(points_sum: float, games: int) -> float | None:
"""Тот же сглаженный рейтинг, что и SMOOTHED_SCORE_SQL, но в Python.
Нужен там, где строки уже вытащены и агрегировать в SQL нечего (профиль игрока).
Держим рядом с SQL-версией и на одних константах: разъехавшиеся реализации одной
формулы — источник расхождений, который потом ловится только глазами."""
if games <= 0:
return None # рейтинга без игр нет — как SUM(...) = NULL в SQL
return None
return (PRIOR_GAMES * PRIOR_MEAN + points_sum) / (PRIOR_GAMES + games) * 100
+197
View File
@@ -0,0 +1,197 @@
"""Движок рейтинга: примеры docs/rating/rating-system.md и сверка с эталоном simulate.py.
Числа примеров — те же, что в документе (раздел 6) и в EXPECTED эталона: разъехаться
документ, эталон и приложение не должны. Сверка с simulate.py дополнительно гоняет
синтетический сезон и требует совпадения каждого изменения рейтинга."""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
import pytest
from app.services import scoring
from app.services.scoring import RatedMatch, RatedSeat, rate_match, replay
VETERAN = 40 # партий у «опытного» игрока: K = K_MIN
A, B, C, D, E, F = 1, 2, 3, 4, 5, 6
def _vets(*ids: int) -> dict[int, int]:
return dict.fromkeys(ids, VETERAN)
def _duel(first: int, second: int, **kw) -> RatedMatch:
return RatedMatch((RatedSeat(first, 1), RatedSeat(second, 2)), **kw)
def _seat(uid: int, place: int, objectives=None, worlds=None, eliminated=False) -> RatedSeat:
return RatedSeat(uid, place, objectives=objectives, worlds=worlds, eliminated=eliminated)
FIVE = tuple(RatedSeat(uid, i + 1) for i, uid in enumerate((A, B, C, D, E)))
SIX = tuple(RatedSeat(uid, i + 1) for i, uid in enumerate((A, B, C, D, E, F)))
# (ключ, рейтинги, сыграно партий, партия, ожидаемые ΔR с точностью до 0.01)
EXAMPLES = [
("1a", {A: 1600, B: 1400}, _vets(A, B), _duel(A, B, win_reason="objectives"),
{A: 3.84, B: -3.84}),
("1b", {A: 1600, B: 1400}, _vets(A, B), _duel(B, A, win_reason="objectives"),
{B: 12.16, A: -12.16}),
("2a", {A: 1500, B: 1500}, _vets(A, B),
RatedMatch((_seat(A, 1, 2, 5), _seat(B, 2, 1, 4)), "objectives", end_round=3),
{A: 11.18, B: -11.18}),
("2b", {A: 1500, B: 1500}, _vets(A, B),
RatedMatch((_seat(A, 1, 2, 5), _seat(B, 2, 1, 4)), "objectives", end_round=8),
{A: 5.46, B: -5.46}),
("3a", {A: 1500, B: 1500}, _vets(A, B), _duel(A, B, win_reason="objectives"),
{A: 8.0, B: -8.0}),
("3b", dict.fromkeys((A, B, C, D, E), 1500), _vets(A, B, C, D, E),
RatedMatch(FIVE, "objectives"),
{A: 11.0, B: 5.5, C: 0.0, D: -5.5, E: -11.0}),
("4a", {A: 1500, B: 1500}, _vets(A, B), _duel(A, B, win_reason="worlds"),
{A: 6.8, B: -6.8}),
("4b", {A: 1500, B: 1500}, _vets(A, B), _duel(A, B, win_reason="plastic"),
{A: 5.6, B: -5.6}),
("4c", {A: 1500, B: 1500}, _vets(A, B), _duel(A, B, win_reason="resources"),
{A: 4.8, B: -4.8}),
("4d", {A: 1500, B: 1500}, _vets(A, B),
RatedMatch((_seat(A, 1, 2, 6), _seat(B, 2, 2, 5)), "worlds", end_round=8),
{A: 3.4, B: -3.4}),
("4e", {A: 1500, B: 1500}, _vets(A, B),
RatedMatch((_seat(A, 1, 2, 8), _seat(B, 2, 0, 2)), "objectives", end_round=3),
{A: 16.0, B: -16.0}),
("5", {A: 1550, B: 1500, C: 1480, D: 1450}, _vets(A, B, C, D),
RatedMatch(
(_seat(A, 1, 4, 8), _seat(B, 2, 3, 7),
_seat(C, 3, 1, 0, eliminated=True), _seat(D, 3, 0, 0, eliminated=True)),
"objectives", end_round=7,
),
{A: 9.27, B: 5.85, C: -7.78, D: -7.35}),
("6a", dict.fromkeys(range(A, F + 1), 1500), _vets(*range(A, F + 1)),
RatedMatch(SIX, "objectives", end_round=8, nine_rounds_rule=True),
{A: 12.0, B: 7.2, C: 2.4, D: -2.4, E: -7.2, F: -12.0}),
("6b", dict.fromkeys(range(A, F + 1), 1500), _vets(*range(A, F + 1)),
RatedMatch(SIX, "objectives", end_round=8, nine_rounds_rule=False),
{A: 10.29, B: 7.54, C: 2.74, D: -2.06, E: -6.86, F: -11.66}),
("7", {A: 1500, B: 1500}, {A: 0, B: VETERAN}, _duel(A, B, win_reason="objectives"),
{A: 32.0, B: -8.0}),
]
@pytest.mark.parametrize(
"ratings,games,match,want", [e[1:] for e in EXAMPLES], ids=[e[0] for e in EXAMPLES]
)
def test_document_examples(ratings, games, match, want):
delta, _perf = rate_match(ratings, games, match)
assert {uid: round(v, 2) for uid, v in delta.items()} == want
def test_examples_cover_whole_section():
assert len(EXAMPLES) == 15
def test_last_standing_counts_full_objective_gap():
"""Победа last_standing: отрыв победителя по целям = 1, сколько бы маркеров ни было."""
seats = (_seat(A, 1, 1, 6), _seat(B, 2, 1, 0, eliminated=True))
ordinary = rate_match({}, _vets(A, B), RatedMatch(seats, "objectives"))[0][A]
standing = rate_match({}, _vets(A, B), RatedMatch(seats, "last_standing"))[0][A]
# Отрыв по целям 1 вместо 0 → множитель больше на W_OBJ·1 = 0.5; ΔR = K·ΔM·(S − E).
assert standing - ordinary == pytest.approx(16 * 0.5 * 0.5)
# ─── Сверка с эталоном ───────────────────────────────────────────────────────
SIMULATE = Path(__file__).resolve().parents[2] / "docs" / "rating" / "simulate.py"
@pytest.fixture(scope="module")
def sim():
if not SIMULATE.exists():
pytest.skip("docs/rating/simulate.py недоступен")
spec = importlib.util.spec_from_file_location("rating_simulate", SIMULATE)
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module # dataclasses ищут модуль по имени
spec.loader.exec_module(module) # type: ignore[union-attr]
return module
def _convert(m, ids: dict[str, int], match_id: int) -> RatedMatch:
return RatedMatch(
tuple(
RatedSeat(
ids[s.player], s.place, eliminated=s.eliminated,
objectives=s.objectives, worlds=s.worlds,
)
for s in m.seats
),
m.win_reason,
end_round=m.round,
nine_rounds_rule=m.nine_rounds,
id=match_id,
)
def test_constants_match_reference(sim):
p = sim.PROPOSED
assert (p.r0, p.d, p.k_max, p.k_min, p.k_games) == (
scoring.R0, scoring.D, scoring.K_MAX, scoring.K_MIN, scoring.K_GAMES
)
assert (p.w_table, p.w_tempo, p.w_obj, p.w_worlds) == (
scoring.W_TABLE, scoring.W_TEMPO, scoring.W_OBJ, scoring.W_WORLDS
)
assert (p.mu_obj, p.mu_worlds, p.m_min, p.m_max) == (
scoring.MU_OBJ, scoring.MU_WORLDS, scoring.M_MIN, scoring.M_MAX
)
assert dict(p.closeness) == scoring.CLOSENESS
assert sim.BOARD_TILES == scoring.BOARD_TILES
assert not p.autocorr
@pytest.mark.parametrize("scenario", ["сигнал", "клубы", "рост"])
@pytest.mark.parametrize("stripped", [False, True], ids=["full", "history"])
def test_replay_matches_reference_season(sim, scenario, stripped):
"""Весь сезон: каждое изменение рейтинга совпадает с эталоном до 1e-9."""
cfg = sim.SCENARIOS[scenario]
_skill, matches = sim.generate_season(
cfg["seed"], sim.SEASON_MATCHES, cfg["informative"], cfg["clubs"], cfg["learning"]
)
if stripped:
matches = [sim.strip_details(m) for m in matches]
ids: dict[str, int] = {}
for m in matches:
for s in m.seats:
ids.setdefault(s.player, len(ids) + 1)
reference = sim.Elo(sim.PROPOSED)
ours = replay(_convert(m, ids, i) for i, m in enumerate(matches))
for i, m in enumerate(matches):
for player, dv in reference.update(m).items():
assert ours.delta[(i, ids[player])] == pytest.approx(dv, abs=1e-9)
for player, uid in ids.items():
assert ours.ratings[uid] == pytest.approx(reference.rating(player), abs=1e-6)
def test_monotone_and_zero_sum(sim):
"""Победитель без ничьей не теряет, последний без ничьей не получает; при равных K
сумма изменений за партию — ноль."""
cfg = sim.SCENARIOS["сигнал"]
_skill, matches = sim.generate_season(cfg["seed"] + 7, 200, True)
ids: dict[str, int] = {}
for m in matches:
for s in m.seats:
ids.setdefault(s.player, len(ids) + 1)
veterans = dict.fromkeys(ids.values(), VETERAN)
for i, m in enumerate(matches):
rm = _convert(m, ids, i)
delta, _ = rate_match({}, veterans, rm)
places = [s.place for s in rm.seats]
for s in rm.seats:
if places.count(s.place) > 1:
continue
if s.place == 1:
assert delta[s.user_id] > 0
if s.place == max(places):
assert delta[s.user_id] < 0
assert sum(delta.values()) == pytest.approx(0.0, abs=1e-9)