#!/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()