NotBigGhostandClaude Opus 5
8f1518b0b3
docs/rating/simulate.py — только stdlib, фиксированные seed. Содержит многопользовательский Elo с множителем отрыва (темп, цели, миры, тип победы, размер стола, K новичка) и пошаговые примеры документа с assert. Синтетическая лига в сценариях «сигнал», «шум», «клубы» и «рост» сравнивает текущий League Points, чистый Elo и предложенную систему. Флаг --grid перебирает K и веса на отдельных сезонах. #22 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XfTsytzT6TojfmprRDKiV6
841 lines
38 KiB
Python
841 lines
38 KiB
Python
#!/usr/bin/env python3
|
||
"""Эталонная реализация и симуляция предложенной рейтинговой системы (#22).
|
||
|
||
Скрипт — приложение к docs/rating/rating-system.md:
|
||
|
||
1. Формулы документа в коде (раздел «Эталонная реализация»). #23 может сверять
|
||
с ними свою реализацию.
|
||
2. Пошаговые примеры документа с assert на числа: документ и код не разъедутся.
|
||
3. Синтетическая лига: игроки со скрытой «истинной» силой, партии на 2–6 человек.
|
||
Детали партии (раунд, цели, миры, тип победы) выводятся из отрыва
|
||
по производительности. На одних и тех же партиях сравниваются текущий League
|
||
Points, чистый парный Elo и предложенная система.
|
||
4. Перебор весов (--grid).
|
||
|
||
Только стандартная библиотека и фиксированные seed — вывод воспроизводим.
|
||
|
||
python docs/rating/simulate.py # примеры + сравнение систем (≈20 с)
|
||
python docs/rating/simulate.py --grid # примеры + перебор K и весов (≈5 мин)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import math
|
||
import random
|
||
import statistics
|
||
import sys
|
||
from dataclasses import dataclass, replace
|
||
from itertools import combinations
|
||
|
||
# ═══ Правила игры ═══════════════════════════════════════════════════════════════
|
||
|
||
# Размер поля в тайлах по числу игроков (6 игроков — 4×5, ответ владельца в #22).
|
||
BOARD_TILES = {2: 4, 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
|
||
|
||
|
||
def worlds_on_board(n: int) -> float:
|
||
return BOARD_TILES[n] * WORLDS_PER_TILE
|
||
|
||
|
||
def fair_worlds(n: int) -> float:
|
||
"""«Честная доля» миров на игрока — масштаб для разницы миров."""
|
||
return worlds_on_board(n) / n
|
||
|
||
|
||
def max_rounds(n: int, nine_rounds: bool) -> int:
|
||
return EXTENDED_ROUNDS if nine_rounds and n >= EXTENDED_MIN_PLAYERS else BASE_ROUNDS
|
||
|
||
|
||
# ═══ Эталонная реализация ═════════════════════════════════════════════════════
|
||
|
||
|
||
@dataclass
|
||
class Seat:
|
||
player: str
|
||
place: int
|
||
objectives: int | None = None # маркеры целей на конец партии
|
||
worlds: int | None = None # дружественные миры на конец партии
|
||
eliminated: bool = False
|
||
|
||
|
||
@dataclass
|
||
class Match:
|
||
seats: list[Seat]
|
||
win_reason: str | None = None
|
||
round: int | None = None # раунд, в котором партия закончилась
|
||
nine_rounds: bool = False # снимок настройки группы на момент партии
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Params:
|
||
# Шкала «Elo / 10»: старт 50, разница 40 = шансы 10:1. Классический Elo — R·10 + 1000.
|
||
r0: float = 50.0 # стартовый рейтинг
|
||
d: float = 40.0 # масштаб логистики
|
||
k_max: float = 6.4 # K новичка (0 партий)
|
||
k_min: float = 1.6 # K опытного игрока
|
||
k_games: int = 20 # за сколько партий K линейно спускается от k_max к k_min
|
||
w_table: float = 0.0 # вес размера стола (треб. 2)
|
||
w_tempo: float = 0.0 # вес темпа победы (треб. 1)
|
||
w_obj: float = 0.0 # вес разницы целей (треб. 4)
|
||
w_worlds: float = 0.0 # вес разницы миров (треб. 4)
|
||
# Близость партии по типу победы (треб. 5) — множитель пар с победителем.
|
||
closeness: tuple[tuple[str, float], ...] = ()
|
||
# «Типичные» значения признаков: партия с ними получает множитель 1,
|
||
# отсутствующий признак подставляется типичным (= нейтральным).
|
||
mu_obj: float = 0.5
|
||
mu_worlds: float = 0.5
|
||
m_min: float = 0.5
|
||
m_max: float = 2.0
|
||
autocorr: bool = False # поправка на автокорреляцию (см. документ)
|
||
|
||
def closeness_for(self, reason: str | None) -> float:
|
||
return dict(self.closeness).get(reason, 1.0) if reason else 1.0
|
||
|
||
|
||
def mu_tempo(rmax: int) -> float:
|
||
"""Типичный темп: партия закончилась в предпоследнем раунде."""
|
||
return 1.0 / (rmax - 1)
|
||
|
||
|
||
def expected(r_a: float, r_b: float, d: float) -> float:
|
||
"""Ожидаемый результат a против b (вероятность, что a окажется выше)."""
|
||
return 1.0 / (1.0 + 10.0 ** ((r_b - r_a) / d))
|
||
|
||
|
||
def k_factor(games: int, p: Params) -> float:
|
||
left = max(0.0, 1.0 - games / p.k_games)
|
||
return p.k_min + (p.k_max - p.k_min) * left
|
||
|
||
|
||
def table_weight(n: int, p: Params) -> float:
|
||
return 1.0 + p.w_table * (n - 2) / 4.0
|
||
|
||
|
||
def _clamp(x: float, lo: float, hi: float) -> float:
|
||
return max(lo, min(hi, x))
|
||
|
||
|
||
def pair_multiplier(
|
||
m: Match, a: Seat, b: Seat, r_a: float, r_b: float, p: Params
|
||
) -> tuple[float, dict]:
|
||
"""Множитель отрыва пары; a — выше или наравне с b. Возвращает (M, разбор)."""
|
||
n = len(m.seats)
|
||
tie = a.place == b.place
|
||
winner_pair = a.place == 1
|
||
parts: dict[str, float] = {}
|
||
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)
|
||
mu = mu_tempo(rmax)
|
||
tempo = mu if m.round is None else (rmax - m.round) / (rmax - 1)
|
||
parts["tempo"] = tempo
|
||
add += p.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 = p.mu_obj
|
||
parts["obj"] = obj
|
||
add += p.w_obj * (obj - p.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 = p.mu_worlds
|
||
parts["worlds"] = wor
|
||
add += p.w_worlds * (wor - p.mu_worlds)
|
||
|
||
if p.autocorr and add > 1.0 and not tie:
|
||
# Поправка FiveThirtyEight (2.2 / (0.001·ΔElo + 2.2)) в шкале Elo/10.
|
||
kappa = 2.2 / (0.01 * (r_a - r_b) + 2.2)
|
||
parts["kappa"] = kappa
|
||
add = 1.0 + (add - 1.0) * kappa
|
||
parts["additive"] = add
|
||
mult = _clamp(add, p.m_min, p.m_max)
|
||
close = p.closeness_for(m.win_reason) if winner_pair else 1.0
|
||
parts["closeness"] = close
|
||
return mult * close, parts
|
||
|
||
|
||
def rate_match(
|
||
ratings: dict[str, float],
|
||
games: dict[str, int],
|
||
m: Match,
|
||
p: Params,
|
||
trace: list | None = None,
|
||
) -> dict[str, float]:
|
||
"""Изменения рейтинга участников партии. Рейтинги/счётчики не мутирует."""
|
||
n = len(m.seats)
|
||
g = table_weight(n, p)
|
||
r = {s.player: ratings.get(s.player, p.r0) for s in m.seats}
|
||
k = {s.player: k_factor(games.get(s.player, 0), p) for s in m.seats}
|
||
delta = {s.player: 0.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.player], r[b.player], p.d)
|
||
mult, parts = pair_multiplier(m, a, b, r[a.player], r[b.player], p)
|
||
x = mult * (s_ab - e_ab)
|
||
delta[a.player] += k[a.player] * g / (n - 1) * x
|
||
delta[b.player] -= k[b.player] * g / (n - 1) * x
|
||
if trace is not None:
|
||
trace.append(
|
||
{"a": a.player, "b": b.player, "S": s_ab, "E": e_ab, "M": mult, **parts}
|
||
)
|
||
return delta
|
||
|
||
|
||
# ═══ Системы для сравнения ════════════════════════════════════════════════════
|
||
|
||
|
||
class LeaguePoints:
|
||
"""Текущая система (backend/app/services/scoring.py): сглаженное среднее очков за место."""
|
||
|
||
probabilistic = False
|
||
PRIOR_GAMES = 10
|
||
PRIOR_MEAN = 0.5
|
||
|
||
def __init__(self) -> None:
|
||
self.sum: dict[str, float] = {}
|
||
self.games: dict[str, int] = {}
|
||
|
||
def rating(self, player: str) -> float:
|
||
g = self.games.get(player, 0)
|
||
return (self.PRIOR_GAMES * self.PRIOR_MEAN + self.sum.get(player, 0.0)) / (
|
||
self.PRIOR_GAMES + g
|
||
) * 100
|
||
|
||
def update(self, m: Match) -> dict[str, float]:
|
||
n = len(m.seats)
|
||
tie = {}
|
||
for s in m.seats:
|
||
tie[s.place] = tie.get(s.place, 0) + 1
|
||
before = {s.player: self.rating(s.player) for s in m.seats}
|
||
for s in m.seats:
|
||
pts = (n - s.place - (tie[s.place] - 1) / 2) / (n - 1)
|
||
self.sum[s.player] = self.sum.get(s.player, 0.0) + pts
|
||
self.games[s.player] = self.games.get(s.player, 0) + 1
|
||
return {s.player: self.rating(s.player) - before[s.player] for s in m.seats}
|
||
|
||
|
||
class Elo:
|
||
"""Парный многопользовательский Elo; с нулевыми весами — «чистый» Elo."""
|
||
|
||
probabilistic = True
|
||
|
||
def __init__(self, p: Params) -> None:
|
||
self.p = p
|
||
self.r: dict[str, float] = {}
|
||
self.games: dict[str, int] = {}
|
||
|
||
def rating(self, player: str) -> float:
|
||
return self.r.get(player, self.p.r0)
|
||
|
||
def update(self, m: Match) -> dict[str, float]:
|
||
delta = rate_match(self.r, self.games, m, self.p)
|
||
for pl, dv in delta.items():
|
||
self.r[pl] = self.rating(pl) + dv
|
||
self.games[pl] = self.games.get(pl, 0) + 1
|
||
return delta
|
||
|
||
|
||
# ═══ Генератор синтетической лиги ═════════════════════════════════════════════
|
||
|
||
SIGMA_SKILL = 150.0 # разброс «истинной» силы игроков
|
||
SIGMA_PERF = 225.0 # шум производительности в отдельной партии (кубы, карты, ошибки)
|
||
# Истинная сила в шкале рейтинга: Φ(Δθ/(σ√2)) ≈ логистика с масштабом d=40 при ΔR ≈ 0.093·Δθ.
|
||
SKILL_TO_RATING = 1.702 * 40 / (math.log(10) * SIGMA_PERF * math.sqrt(2))
|
||
TABLE_SIZES = ((2, 0.45), (3, 0.25), (4, 0.20), (5, 0.07), (6, 0.03))
|
||
ELIM_Z = 2.3 # отставание (в σ), при котором игрок может выбыть
|
||
ELIM_P = 0.35 # вероятность выбывания при таком отставании
|
||
|
||
|
||
def _choice_weighted(rng: random.Random, pairs) -> int:
|
||
x = rng.random() * sum(w for _, w in pairs)
|
||
for v, w in pairs:
|
||
x -= w
|
||
if x <= 0:
|
||
return v
|
||
return pairs[-1][0]
|
||
|
||
|
||
def generate_match(
|
||
rng: random.Random,
|
||
skill: dict[str, float],
|
||
players: list[str],
|
||
nine_rounds: bool,
|
||
informative: bool,
|
||
) -> Match:
|
||
"""Партия: места — по производительности, детали — по отрыву.
|
||
|
||
informative=False — сценарий «шум»: места те же, но величина отрыва (а значит
|
||
раунд, цели, миры и тип победы) не связана с силой игроков."""
|
||
n = len(players)
|
||
perf = {pl: skill[pl] + rng.gauss(0, SIGMA_PERF) for pl in players}
|
||
order = sorted(players, key=perf.get, reverse=True)
|
||
if informative:
|
||
z = {pl: (perf[order[0]] - perf[pl]) / SIGMA_PERF for pl in players}
|
||
else:
|
||
ghost = sorted((rng.gauss(0, SIGMA_PERF) for _ in players), reverse=True)
|
||
z = {pl: (ghost[0] - ghost[i]) / SIGMA_PERF for i, pl in enumerate(order)}
|
||
|
||
winner = order[0]
|
||
eliminated = {pl for pl in order[1:] if z[pl] > ELIM_Z and rng.random() < ELIM_P}
|
||
survivors = [pl for pl in order if pl not in eliminated]
|
||
rmax = max_rounds(n, nine_rounds)
|
||
|
||
if len(survivors) == 1:
|
||
reason = "last_standing"
|
||
gap = z[order[1]]
|
||
else:
|
||
gap = z[survivors[1]]
|
||
if gap < 0.02:
|
||
reason = "resources"
|
||
elif gap < 0.08:
|
||
reason = "plastic"
|
||
elif gap < 0.25:
|
||
reason = "worlds"
|
||
else:
|
||
reason = "objectives"
|
||
rnd = int(_clamp(round(rmax + 0.3 - 1.4 * gap + rng.gauss(0, 0.9)), 3, rmax))
|
||
|
||
need = n
|
||
if reason == "last_standing":
|
||
o_win = rng.randint(max(0, need - 2), need - 1)
|
||
elif rnd < rmax:
|
||
o_win = need
|
||
else:
|
||
o_win = need if rng.random() < 0.5 else need - 1
|
||
objectives = {winner: o_win}
|
||
for pl in order[1:]:
|
||
o = round(o_win * (1 - 0.45 * z[pl]) + rng.gauss(0, 0.5))
|
||
objectives[pl] = int(_clamp(o, 0, max(0, o_win - 1)))
|
||
|
||
total = int(worlds_on_board(n))
|
||
mean_z = statistics.fmean(z[pl] for pl in survivors)
|
||
worlds = {}
|
||
for pl in order:
|
||
if pl in eliminated:
|
||
worlds[pl] = 0
|
||
continue
|
||
w = round(fair_worlds(n) * (1 + 0.35 * (mean_z - z[pl])) + rng.gauss(0, 0.8))
|
||
worlds[pl] = int(_clamp(w, 1, total))
|
||
|
||
if reason in ("worlds", "plastic", "resources"):
|
||
ru = survivors[1]
|
||
objectives[ru] = o_win
|
||
if reason == "worlds":
|
||
if worlds[winner] <= worlds[ru]:
|
||
worlds[winner] = worlds[ru] + 1
|
||
else:
|
||
worlds[ru] = worlds[winner]
|
||
|
||
seats = []
|
||
for i, pl in enumerate(survivors):
|
||
seats.append(Seat(pl, i + 1, objectives[pl], worlds[pl]))
|
||
last = len(survivors) + 1
|
||
for pl in order:
|
||
if pl in eliminated:
|
||
seats.append(Seat(pl, last, objectives[pl], 0, eliminated=True))
|
||
return Match(seats, reason, rnd, nine_rounds)
|
||
|
||
|
||
def strip_details(m: Match) -> Match:
|
||
"""Партия «из старой истории»: только места и тип победы."""
|
||
return Match(
|
||
[Seat(s.player, s.place, eliminated=s.eliminated) for s in m.seats], m.win_reason
|
||
)
|
||
|
||
|
||
CLUB_OFFSETS = (-150.0, 0.0, 150.0) # сценарий «клубы»: средняя сила трёх групп
|
||
CLUB_SIGMA = 90.0 # разброс силы внутри клуба
|
||
CLUB_MIXED_SHARE = 0.1 # доля партий, где встречаются игроки разных клубов
|
||
|
||
|
||
LEARN_DEFICIT = 200.0 # сценарий «рост»: максимальное стартовое отставание новичка
|
||
LEARN_GAMES = 15.0 # за столько партий отставание уменьшается в e раз
|
||
|
||
|
||
def generate_season(
|
||
seed: int, n_matches: int, informative: bool, clubs: bool = False, learning: bool = False
|
||
) -> tuple[dict, list[Match]]:
|
||
"""Сезон: 12 игроков сразу, ещё по двое на 1/3 и 2/3 сезона.
|
||
|
||
clubs=True — игроки разбиты на три группы разной силы и почти всегда играют
|
||
внутри своей; общий рейтинг должен их правильно «сшить».
|
||
learning=True — сила растёт с опытом: θ − deficit·exp(−партии/LEARN_GAMES).
|
||
Возвращает силу на КОНЕЦ сезона — её и должен отражать рейтинг."""
|
||
rng = random.Random(seed)
|
||
skill = {}
|
||
activity = {}
|
||
joins = {}
|
||
club = {}
|
||
deficit = {}
|
||
played = {}
|
||
for i in range(18 if clubs else 16):
|
||
pl = f"p{i:02d}"
|
||
if clubs:
|
||
club[pl] = i % 3
|
||
skill[pl] = CLUB_OFFSETS[club[pl]] + rng.gauss(0, CLUB_SIGMA)
|
||
joins[pl] = 0 if i < 15 else n_matches // 3
|
||
else:
|
||
skill[pl] = rng.gauss(0, SIGMA_SKILL)
|
||
joins[pl] = 0 if i < 12 else (n_matches // 3 if i < 14 else 2 * n_matches // 3)
|
||
activity[pl] = rng.uniform(0.5, 1.5)
|
||
deficit[pl] = rng.uniform(0, LEARN_DEFICIT) if learning else 0.0
|
||
played[pl] = 0
|
||
|
||
def current(pl: str) -> float:
|
||
return skill[pl] - deficit[pl] * math.exp(-played[pl] / LEARN_GAMES)
|
||
|
||
nine_rounds = rng.random() < 0.5
|
||
matches = []
|
||
for t in range(n_matches):
|
||
active = [pl for pl in skill if joins[pl] <= t]
|
||
if clubs and rng.random() >= CLUB_MIXED_SHARE:
|
||
c = rng.randrange(3)
|
||
active = [pl for pl in active if club[pl] == c]
|
||
n = min(_choice_weighted(rng, TABLE_SIZES), len(active))
|
||
pool = active[:]
|
||
chosen = []
|
||
for _ in range(n):
|
||
pick = _choice_weighted(rng, [(pl, activity[pl]) for pl in pool])
|
||
pool.remove(pick)
|
||
chosen.append(pick)
|
||
now = {pl: current(pl) for pl in chosen}
|
||
matches.append(generate_match(rng, now, chosen, nine_rounds, informative))
|
||
for pl in chosen:
|
||
played[pl] += 1
|
||
return {pl: current(pl) for pl in skill}, matches
|
||
|
||
|
||
# ═══ Метрики ══════════════════════════════════════════════════════════════════
|
||
|
||
|
||
def _ranks(xs: list[float]) -> list[float]:
|
||
order = sorted(range(len(xs)), key=lambda i: xs[i])
|
||
ranks = [0.0] * len(xs)
|
||
i = 0
|
||
while i < len(order):
|
||
j = i
|
||
while j + 1 < len(order) and xs[order[j + 1]] == xs[order[i]]:
|
||
j += 1
|
||
for t in range(i, j + 1):
|
||
ranks[order[t]] = (i + j) / 2 + 1
|
||
i = j + 1
|
||
return ranks
|
||
|
||
|
||
def spearman(xs: list[float], ys: list[float]) -> float:
|
||
if len(xs) < 3:
|
||
return float("nan")
|
||
rx, ry = _ranks(xs), _ranks(ys)
|
||
return statistics.correlation(rx, ry)
|
||
|
||
|
||
CHECKPOINTS = (5, 10, 20)
|
||
|
||
|
||
def evaluate(system, skill: dict[str, float], matches: list[Match], feed=None) -> dict:
|
||
"""Прогоняет сезон. feed(m) — какую версию партии видит система (по умолчанию полную)."""
|
||
half = len(matches) // 2
|
||
hits = pairs = 0.0
|
||
brier = []
|
||
abs_moves = []
|
||
at_k: dict[int, dict[str, float]] = {k: {} for k in CHECKPOINTS}
|
||
games: dict[str, int] = {}
|
||
for t, m in enumerate(matches):
|
||
seen = feed(m) if feed else m
|
||
if t >= half:
|
||
for a, b in combinations(m.seats, 2):
|
||
if a.place == b.place:
|
||
continue
|
||
if a.place > b.place:
|
||
a, b = b, a
|
||
ra, rb = system.rating(a.player), system.rating(b.player)
|
||
pairs += 1
|
||
hits += 1.0 if ra > rb else 0.5 if ra == rb else 0.0
|
||
if system.probabilistic:
|
||
brier.append((1.0 - expected(ra, rb, system.p.d)) ** 2)
|
||
delta = system.update(seen)
|
||
for s in m.seats:
|
||
games[s.player] = games.get(s.player, 0) + 1
|
||
gp = games[s.player]
|
||
if gp > 20:
|
||
abs_moves.append(abs(delta[s.player]))
|
||
if gp in at_k:
|
||
at_k[gp][s.player] = system.rating(s.player)
|
||
played = [pl for pl in skill if games.get(pl, 0) >= 10]
|
||
out = {
|
||
"acc": hits / pairs if pairs else float("nan"),
|
||
"rho": spearman([system.rating(pl) for pl in played], [skill[pl] for pl in played]),
|
||
"move": statistics.fmean(abs_moves) if abs_moves else float("nan"),
|
||
}
|
||
for k in CHECKPOINTS:
|
||
pls = list(at_k[k])
|
||
out[f"rho@{k}"] = spearman([at_k[k][pl] for pl in pls], [skill[pl] for pl in pls])
|
||
if system.probabilistic:
|
||
out["brier"] = statistics.fmean(brier)
|
||
rs = [system.rating(pl) for pl in played]
|
||
ts = [skill[pl] * SKILL_TO_RATING for pl in played]
|
||
mr, mt = statistics.fmean(rs), statistics.fmean(ts)
|
||
out["rmse"] = math.sqrt(statistics.fmean(((r - mr) - (t - mt)) ** 2 for r, t in zip(rs, ts)))
|
||
everyone = [system.rating(pl) for pl in games]
|
||
out["inflation"] = statistics.fmean(everyone) - system.p.r0
|
||
out["slope"] = _slope(ts, rs)
|
||
return out
|
||
|
||
|
||
def oracle_accuracy(skill: dict[str, float], matches: list[Match]) -> float:
|
||
half = len(matches) // 2
|
||
hits = pairs = 0
|
||
for m in matches[half:]:
|
||
for a, b in combinations(m.seats, 2):
|
||
if a.place == b.place:
|
||
continue
|
||
if a.place > b.place:
|
||
a, b = b, a
|
||
pairs += 1
|
||
hits += skill[a.player] > skill[b.player]
|
||
return hits / pairs
|
||
|
||
|
||
def _slope(xs: list[float], ys: list[float]) -> float:
|
||
"""Наклон регрессии рейтинга на истинную силу: 1 — масштаб честный, >1 — раздут."""
|
||
mx, my = statistics.fmean(xs), statistics.fmean(ys)
|
||
sxx = sum((x - mx) ** 2 for x in xs)
|
||
return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sxx if sxx else float("nan")
|
||
|
||
|
||
def summarize(rows: list[dict]) -> dict[str, tuple[float, float]]:
|
||
keys = rows[0].keys()
|
||
res = {}
|
||
for k in keys:
|
||
vals = [r[k] for r in rows if not math.isnan(r[k])]
|
||
mean = statistics.fmean(vals)
|
||
se = statistics.stdev(vals) / math.sqrt(len(vals)) if len(vals) > 1 else 0.0
|
||
res[k] = (mean, se)
|
||
return res
|
||
|
||
|
||
# ═══ Коэффициенты ═════════════════════════════════════════════════════════════
|
||
|
||
PLAIN = Params(k_max=9.6, k_min=1.6) # чистый Elo со своими лучшими K (перебор, этап 1)
|
||
CLOSENESS = (
|
||
("objectives", 1.0),
|
||
("worlds", 0.85),
|
||
("plastic", 0.7),
|
||
("resources", 0.6),
|
||
("last_standing", 1.0),
|
||
)
|
||
# Для анализа чувствительности: отклонения от 1 вдвое больше.
|
||
CLOSENESS_STRONG = tuple((r, 1.0 - 2 * (1.0 - c)) for r, c in CLOSENESS)
|
||
PROPOSED = Params(
|
||
w_table=0.5,
|
||
w_tempo=1.0,
|
||
w_obj=0.5,
|
||
w_worlds=0.5,
|
||
closeness=CLOSENESS,
|
||
)
|
||
|
||
|
||
# ═══ Примеры из документа ═════════════════════════════════════════════════════
|
||
|
||
VETERAN = 40 # партий у «опытного» игрока: K = k_min
|
||
|
||
|
||
def _vets(*names: str) -> dict[str, int]:
|
||
return {n: VETERAN for n in names}
|
||
|
||
|
||
def examples() -> list[tuple[str, str, dict, dict, Match]]:
|
||
"""(ключ, заголовок, рейтинги, сыграно партий, партия) — в порядке документа."""
|
||
duel = lambda first, second, **kw: Match([Seat(first, 1), Seat(second, 2)], **kw) # noqa: E731
|
||
five = [Seat("A", 1), Seat("B", 2), Seat("C", 3), Seat("D", 4), Seat("E", 5)]
|
||
six = [Seat(x, i + 1) for i, x in enumerate("ABCDEF")]
|
||
return [
|
||
("1a", "Дуэль 60 против 40: побеждает сильный",
|
||
{"A": 60, "B": 40}, _vets("A", "B"), duel("A", "B", win_reason="objectives")),
|
||
("1b", "Дуэль 60 против 40: побеждает слабый",
|
||
{"A": 60, "B": 40}, _vets("A", "B"), duel("B", "A", win_reason="objectives")),
|
||
("2a", "Быстрая победа: 3-й раунд",
|
||
{"A": 50, "B": 50}, _vets("A", "B"),
|
||
Match([Seat("A", 1, 2, 5), Seat("B", 2, 1, 4)], "objectives", round=3)),
|
||
("2b", "Медленная победа: 8-й раунд",
|
||
{"A": 50, "B": 50}, _vets("A", "B"),
|
||
Match([Seat("A", 1, 2, 5), Seat("B", 2, 1, 4)], "objectives", round=8)),
|
||
("3a", "Первое место в дуэли",
|
||
{"A": 50, "B": 50}, _vets("A", "B"), duel("A", "B", win_reason="objectives")),
|
||
("3b", "Стол на 5: все места",
|
||
dict.fromkeys("ABCDE", 50), _vets(*"ABCDE"), Match(five, "objectives")),
|
||
("4a", "Тип победы без деталей: по мирам",
|
||
{"A": 50, "B": 50}, _vets("A", "B"), duel("A", "B", win_reason="worlds")),
|
||
("4b", "Тип победы без деталей: по пластику",
|
||
{"A": 50, "B": 50}, _vets("A", "B"), duel("A", "B", win_reason="plastic")),
|
||
("4c", "Тип победы без деталей: по ресурсам",
|
||
{"A": 50, "B": 50}, _vets("A", "B"), duel("A", "B", win_reason="resources")),
|
||
("4d", "Самая близкая полная партия: по мирам на 8-м раунде, 2:2 цели, 6:5 миров",
|
||
{"A": 50, "B": 50}, _vets("A", "B"),
|
||
Match([Seat("A", 1, 2, 6), Seat("B", 2, 2, 5)], "worlds", round=8)),
|
||
("4e", "Разгром: 3-й раунд, 2:0 цели, 8:2 миров",
|
||
{"A": 50, "B": 50}, _vets("A", "B"),
|
||
Match([Seat("A", 1, 2, 8), Seat("B", 2, 0, 2)], "objectives", round=3)),
|
||
("5", "Стол на 4: ничья выбывших, раунд 7",
|
||
{"A": 55, "B": 50, "C": 48, "D": 45}, _vets(*"ABCD"),
|
||
Match(
|
||
[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", round=7,
|
||
)),
|
||
("6a", "Стол на 6, конец на 8-м раунде, хоумрул 9 раундов включён",
|
||
dict.fromkeys("ABCDEF", 50), _vets(*"ABCDEF"),
|
||
Match(six, "objectives", round=8, nine_rounds=True)),
|
||
("6b", "Стол на 6, конец на 8-м раунде, хоумрул выключен",
|
||
dict.fromkeys("ABCDEF", 50), _vets(*"ABCDEF"),
|
||
Match(six, "objectives", round=8, nine_rounds=False)),
|
||
("7", "Новичок (0 партий) побеждает ветерана, оба 50",
|
||
{"A": 50, "B": 50}, {"A": 0, "B": VETERAN}, duel("A", "B", win_reason="objectives")),
|
||
]
|
||
|
||
|
||
# Изменения рейтинга в примерах (округление до 0.001) — те же числа стоят в документе.
|
||
EXPECTED: dict[str, dict[str, float]] = {
|
||
"1a": {"A": 0.384, "B": -0.384},
|
||
"1b": {"B": 1.216, "A": -1.216},
|
||
"2a": {"A": 1.148, "B": -1.148},
|
||
"2b": {"A": 0.577, "B": -0.577},
|
||
"3a": {"A": 0.8, "B": -0.8},
|
||
"3b": {"A": 1.1, "B": 0.55, "C": 0.0, "D": -0.55, "E": -1.1},
|
||
"4a": {"A": 0.68, "B": -0.68},
|
||
"4b": {"A": 0.56, "B": -0.56},
|
||
"4c": {"A": 0.48, "B": -0.48},
|
||
"4d": {"A": 0.34, "B": -0.34},
|
||
"4e": {"A": 1.6, "B": -1.6},
|
||
"5": {"A": 0.927, "B": 0.585, "C": -0.778, "D": -0.735},
|
||
"6a": {"A": 1.2, "B": 0.72, "C": 0.24, "D": -0.24, "E": -0.72, "F": -1.2},
|
||
"6b": {"A": 1.029, "B": 0.754, "C": 0.274, "D": -0.206, "E": -0.686, "F": -1.166},
|
||
"7": {"A": 3.2, "B": -0.8},
|
||
}
|
||
|
||
|
||
def run_examples(p: Params = PROPOSED, verbose: bool = True) -> None:
|
||
for key, title, ratings, games, m in examples():
|
||
trace: list = []
|
||
delta = rate_match(ratings, games, m, p, trace)
|
||
got = {pl: round(v, 3) for pl, v in delta.items()}
|
||
if verbose:
|
||
n = len(m.seats)
|
||
print(f"\n### Пример {key}. {title}\n")
|
||
print(f"N={n}, G(N)={table_weight(n, p):.3f}, раунд={m.round}, R_max={max_rounds(n, m.nine_rounds)}, "
|
||
f"тип={m.win_reason}, K: " + ", ".join(f"{pl}={k_factor(games[pl], p):.1f}" for pl in ratings))
|
||
print("\n| пара | S | E | темп | цели | миры | сумма | близость | M |")
|
||
print("|---|---|---|---|---|---|---|---|---|")
|
||
for t in trace:
|
||
tempo = f"{t['tempo']:.3f}" if "tempo" in t else "—"
|
||
print(f"| {t['a']}–{t['b']} | {t['S']} | {t['E']:.3f} | {tempo} | {t['obj']:.3f} | "
|
||
f"{t['worlds']:.3f} | {t['additive']:.3f} | {t['closeness']} | {t['M']:.3f} |")
|
||
print("\nΔR: " + ", ".join(f"{pl} {v:+.3f}" for pl, v in got.items()))
|
||
if EXPECTED:
|
||
assert got == EXPECTED[key], f"пример {key}: {got} ≠ {EXPECTED[key]}"
|
||
if EXPECTED and verbose:
|
||
print("\nВсе примеры совпадают с документом.")
|
||
|
||
|
||
# ═══ Сценарии запуска ═════════════════════════════════════════════════════════
|
||
|
||
SEASON_MATCHES = 300
|
||
|
||
|
||
SCENARIOS = {
|
||
"сигнал": {"informative": True, "clubs": False, "learning": False, "seed": 10_000},
|
||
"шум": {"informative": False, "clubs": False, "learning": False, "seed": 30_000},
|
||
"клубы": {"informative": True, "clubs": True, "learning": False, "seed": 40_000},
|
||
"рост": {"informative": True, "clubs": False, "learning": True, "seed": 50_000},
|
||
}
|
||
|
||
|
||
def compare(seasons: int, scenario: str, proposed: Params) -> None:
|
||
cfg = SCENARIOS[scenario]
|
||
print(f"\n## Сравнение систем — сценарий «{scenario}», {seasons} сезонов по {SEASON_MATCHES} партий\n")
|
||
variants = [
|
||
("League Points (сейчас)", lambda: LeaguePoints(), None),
|
||
("Elo, чистый", lambda: Elo(PLAIN), None),
|
||
("Предложенная", lambda: Elo(proposed), None),
|
||
("Предложенная, без новых полей", lambda: Elo(proposed), strip_details),
|
||
]
|
||
results = {name: [] for name, _, _ in variants}
|
||
oracle = []
|
||
for s in range(seasons):
|
||
skill, matches = generate_season(
|
||
cfg["seed"] + s, SEASON_MATCHES, cfg["informative"], cfg["clubs"], cfg["learning"]
|
||
)
|
||
oracle.append(oracle_accuracy(skill, matches))
|
||
for name, make, feed in variants:
|
||
results[name].append(evaluate(make(), skill, matches, feed))
|
||
print(f"Потолок точности (прогноз по истинной силе): {statistics.fmean(oracle):.4f}\n")
|
||
cols = ["acc", "brier", "rho", "rho@5", "rho@10", "rho@20", "rmse", "slope", "move", "inflation"]
|
||
print("| Система | " + " | ".join(cols) + " |")
|
||
print("|---" * (len(cols) + 1) + "|")
|
||
for name, _, _ in variants:
|
||
sm = summarize(results[name])
|
||
cells = []
|
||
for c in cols:
|
||
if c not in sm:
|
||
cells.append("—")
|
||
else:
|
||
mean, se = sm[c]
|
||
cells.append(f"{mean:.4f} ±{se:.4f}" if c in ("acc", "brier") else f"{mean:.3f}")
|
||
print(f"| {name} | " + " | ".join(cells) + " |")
|
||
base = results["Elo, чистый"]
|
||
prop = results["Предложенная"]
|
||
lp = results["League Points (сейчас)"]
|
||
d_brier = [p["brier"] - b["brier"] for p, b in zip(prop, base)]
|
||
d_acc_lp = [p["acc"] - b["acc"] for p, b in zip(prop, lp)]
|
||
d_acc = [p["acc"] - b["acc"] for p, b in zip(prop, base)]
|
||
for title, ds in (
|
||
("Brier: предложенная − чистый Elo", d_brier),
|
||
("Точность: предложенная − чистый Elo", d_acc),
|
||
("Точность: предложенная − League Points", d_acc_lp),
|
||
):
|
||
mean = statistics.fmean(ds)
|
||
se = statistics.stdev(ds) / math.sqrt(len(ds))
|
||
print(f"- {title}: {mean:+.4f} ± {se:.4f} (парная разница)")
|
||
|
||
|
||
GRID_SEED_SHIFT = 100_000 # перебор идёт на других сезонах, чем итоговое сравнение
|
||
|
||
|
||
def grid(seasons: int) -> None:
|
||
"""Подбор K и весов по Brier (меньше — лучше) на отдельных от сравнения сезонах."""
|
||
data: dict[str, list] = {}
|
||
for name, cfg in SCENARIOS.items():
|
||
data[name] = [
|
||
generate_season(
|
||
cfg["seed"] + GRID_SEED_SHIFT + s, SEASON_MATCHES,
|
||
cfg["informative"], cfg["clubs"], cfg["learning"],
|
||
)
|
||
for s in range(seasons)
|
||
]
|
||
signal = [n for n, c in SCENARIOS.items() if c["informative"]]
|
||
cache: dict = {}
|
||
|
||
def run(p: Params, scenario: str) -> tuple[float, float]:
|
||
if (p, scenario) not in cache:
|
||
rows = [evaluate(Elo(p), sk, ms) for sk, ms in data[scenario]]
|
||
cache[(p, scenario)] = (
|
||
statistics.fmean(r["brier"] for r in rows),
|
||
statistics.fmean(r["rho@10"] for r in rows),
|
||
)
|
||
return cache[(p, scenario)]
|
||
|
||
def brier(p: Params, scenarios) -> float:
|
||
return statistics.fmean(run(p, sc)[0] for sc in scenarios)
|
||
|
||
def rho10(p: Params, scenarios) -> float:
|
||
return statistics.fmean(run(p, sc)[1] for sc in scenarios)
|
||
|
||
print(f"\n## Перебор: {seasons} сезонов на сценарий, критерий — средний Brier\n")
|
||
print("### Этап 1. K чистого Elo (все сценарии)\n")
|
||
print("| k_max | k_min | k_games | Brier |")
|
||
print("|---|---|---|---|")
|
||
k_res = []
|
||
for k_games in (10, 20):
|
||
for k_max in (4.8, 6.4, 9.6, 12.8, 16.0):
|
||
for k_min in (1.6, 2.4, 3.2):
|
||
p = replace(PLAIN, k_max=k_max, k_min=k_min, k_games=k_games)
|
||
b = brier(p, SCENARIOS)
|
||
k_res.append((b, k_max, k_min, k_games))
|
||
print(f"| {k_max} | {k_min} | {k_games} | {b:.5f} |")
|
||
_, k_max, k_min, k_games = min(k_res)
|
||
print(f"\nЛучшие K: k_max={k_max}, k_min={k_min}, k_games={k_games}")
|
||
|
||
print(f"\n### Этап 2. Веса отрыва (K этапа 1; сценарии {', '.join(signal)}; «шум» — контроль)\n")
|
||
res = []
|
||
for w_table in (0.0, 0.25, 0.5):
|
||
for w_tempo in (0.0, 0.5, 1.0):
|
||
for w_obj in (0.0, 0.5, 1.0):
|
||
for w_worlds in (0.0, 0.5, 1.0):
|
||
for close in ((), CLOSENESS):
|
||
p = Params(
|
||
k_max=k_max, k_min=k_min, k_games=k_games, w_table=w_table,
|
||
w_tempo=w_tempo, w_obj=w_obj, w_worlds=w_worlds, closeness=close,
|
||
)
|
||
res.append((brier(p, signal), p))
|
||
res.sort(key=lambda x: x[0])
|
||
zero = next(b for b, p in res if (p.w_table, p.w_tempo, p.w_obj, p.w_worlds) == (0, 0, 0, 0) and not p.closeness)
|
||
plain_best = replace(PLAIN, k_max=k_max, k_min=k_min, k_games=k_games)
|
||
print(f"Без множителя (все веса 0, без близости): Brier {zero:.5f}, "
|
||
f"«шум» {brier(plain_best, ['шум']):.5f}\n")
|
||
print("| Brier | Brier «шум» | w_table | w_tempo | w_obj | w_worlds | близость |")
|
||
print("|---|---|---|---|---|---|---|")
|
||
for b, p in res[:12]:
|
||
noise = brier(p, ["шум"])
|
||
print(
|
||
f"| {b:.5f} | {noise:.5f} | {p.w_table} | {p.w_tempo} | {p.w_obj} | {p.w_worlds} | "
|
||
f"{'да' if p.closeness else 'нет'} |"
|
||
)
|
||
print(f"\nХудшая комбинация: Brier {res[-1][0]:.5f}")
|
||
|
||
print("\n### Этап 3. Доводка K и поправка на автокорреляцию для предложенных весов\n")
|
||
print("| k_max | k_min | k_games | autocorr | Brier (сигнальные) | Brier «шум» |")
|
||
print("|---|---|---|---|---|---|")
|
||
for k_games2 in (10, 20):
|
||
for k_max2 in (4.8, 6.4, 8.0):
|
||
for k_min2 in (1.2, 1.6, 2.4):
|
||
for ac in (False, True):
|
||
p = replace(PROPOSED, k_max=k_max2, k_min=k_min2, k_games=k_games2, autocorr=ac)
|
||
print(
|
||
f"| {k_max2} | {k_min2} | {k_games2} | {'да' if ac else 'нет'} | "
|
||
f"{brier(p, signal):.5f} | {brier(p, ['шум']):.5f} |"
|
||
)
|
||
|
||
print("\n### Этап 4. Чувствительность: один вес меняется, остальные — как в PROPOSED\n")
|
||
print(f"PROPOSED: Brier {brier(PROPOSED, signal):.5f}, ρ@10 {rho10(PROPOSED, signal):.3f}, "
|
||
f"Brier «шум» {brier(PROPOSED, ['шум']):.5f}\n")
|
||
print("| параметр | значение | Brier (сигнальные) | ρ@10 (сигнальные) | Brier «шум» |")
|
||
print("|---|---|---|---|---|")
|
||
sweeps = [
|
||
("w_table", (0.0, 0.25, 0.5, 1.0)),
|
||
("w_tempo", (0.0, 0.25, 0.5, 1.0, 2.0)),
|
||
("w_obj", (0.0, 0.25, 0.5, 1.0, 2.0)),
|
||
("w_worlds", (0.0, 0.25, 0.5, 1.0, 2.0)),
|
||
]
|
||
for attr, values in sweeps:
|
||
for v in values:
|
||
p = replace(PROPOSED, **{attr: v})
|
||
print(f"| {attr} | {v} | {brier(p, signal):.5f} | {rho10(p, signal):.3f} | {brier(p, ['шум']):.5f} |")
|
||
for label, close in (("без близости", ()), ("близость ×2 сильнее", CLOSENESS_STRONG), ("предложенная", CLOSENESS)):
|
||
p = replace(PROPOSED, closeness=close)
|
||
print(f"| closeness | {label} | {brier(p, signal):.5f} | {rho10(p, signal):.3f} | {brier(p, ['шум']):.5f} |")
|
||
|
||
|
||
def main() -> None:
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("--seasons", type=int, default=200, help="сезонов в сравнении систем")
|
||
ap.add_argument("--grid", action="store_true", help="перебор K и весов")
|
||
ap.add_argument("--grid-seasons", type=int, default=40)
|
||
args = ap.parse_args()
|
||
print("# Примеры расчётов")
|
||
run_examples()
|
||
if args.grid:
|
||
grid(args.grid_seasons)
|
||
return
|
||
for scenario in SCENARIOS:
|
||
compare(args.seasons, scenario, PROPOSED)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|