diff --git a/docs/rating/simulate.py b/docs/rating/simulate.py new file mode 100644 index 0000000..3ed973f --- /dev/null +++ b/docs/rating/simulate.py @@ -0,0 +1,840 @@ +#!/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()