Рейтинг: витрины статистики на проигрывании истории

SQL-агрегат League Points (SCORED_CTE, сглаженное среднее) заменён
проигрыванием завершённых партий по порядку played_at, finished_at, id:
load_history — один запрос на партии с участниками, scoring.replay считает
рейтинг. Цепочки две: общая (все партии) и групповая (только партии группы,
K — по партиям внутри группы). Правка или удаление прошлой партии
пересчитывает всё после неё без отдельной логики.

Топ, профиль, главная и статистика группы берут игры, победы, винрейт
и среднее место из той же истории; score — рейтинг целым числом, сортировка
по неокруглённому. Главная грузит историю один раз. Лучшая партия — наибольший
ΔR в общей цепочке, лучшая/худшая фракция — средний S − E. League Points
из scoring.py удалён.

Тесты: дуэль новичков ±32, ранжирование после 10 партий, пересчёт после
правки прошлой партии, отдельная групповая цепочка, лучшая партия по ΔR,
одна загрузка истории на главной. #23

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LqSoRj99iwVEH5U5fnZgsd

backend/app/schemas/api.py

@@ -375,7 +375,8 @@ class OverallStats(BaseModel):
     wins: int
     win_rate: float
     avg_place: float | None = None
-    score: float | None = None
+    # Рейтинг (Elo, старт 1500) целым числом; None — игрок ещё не сыграл ни одной партии.
+    score: int | None = None
 
 
 class LeaderboardEntry(OverallStats):
@@ -416,6 +417,8 @@ class FactionStat(BaseModel):
     wins: int
     win_rate: float
     avg_place: float | None = None
+    # Средний результат относительно ожидания (S − E) × 100 — метрика лучшей/худшей
+    # фракции: выше нуля — игрок на ней выступает лучше своих рейтинговых шансов.
     score: float | None = None
 
 

backend/app/services/scoring.py

@@ -203,33 +203,13 @@ def replay(matches: Iterable[RatedMatch]) -> Replay:
     return out
 
 
-# ─── League Points (прежняя система; уходит вместе с SQL-витринами) ──────────
-
-PRIOR_GAMES = 10
-PRIOR_MEAN = 0.5
-
-SMOOTHED_SCORE_SQL = (
-    f"({PRIOR_GAMES} * {PRIOR_MEAN} + SUM(s.points)) / ({PRIOR_GAMES} + COUNT(*)) * 100"
-)
-
-MATCH_POINTS_SQL = (
-    "CASE WHEN m.player_count > 1 "
-    "THEN (m.player_count - mp.place - (t.tie_size - 1) / 2.0) "
-    "/ (m.player_count - 1) "
-    "ELSE 1.0 END"
-)
-
-
-def smoothed_score(points_sum: float, games: int) -> float | None:
-    if games <= 0:
-        return None
-    return (PRIOR_GAMES * PRIOR_MEAN + points_sum) / (PRIOR_GAMES + games) * 100
-
-
 def leaderboard_sort_key(row: dict) -> tuple:
-    """Ключ сортировки топа: счёт ↓, winrate ↓, игры ↓, среднее место ↑, ник ↑."""
+    """Ключ сортировки топа: рейтинг ↓, winrate ↓, игры ↓, среднее место ↑, ник ↑.
+
+    row["rating"] — рейтинг без округления: два игрока с одинаковым целым в топе
+    всё равно упорядочены по настоящему значению."""
     return (
-        -(row["score"] or 0.0),
+        -(row["rating"] or 0.0),
         -(row["win_rate"] or 0.0),
         -(row["games"] or 0),
         (row["avg_place"] or 0.0),

backend/app/services/stats_service.py

@@ -1,105 +1,151 @@
-"""Статистика и рейтинги. Считается «вживую» (объём данных мал, кэш не нужен)."""
+"""Статистика и рейтинги. Считается «вживую» (объём данных мал, кэш не нужен).
+
+Рейтинг — функция упорядоченной истории (scoring.replay), поэтому витрины не агрегируют
+SQL, а проигрывают завершённые партии: одна загрузка истории на запрос, из неё же
+считаются игры, победы, среднее место и разбивки. Цепочек две: общая — по всем партиям
+приложения, групповая — только по партиям группы (это разные числа)."""
 from __future__ import annotations
 
-from typing import Any
+from collections import defaultdict
 
-from sqlalchemy import func, text
+from sqlalchemy import func
 from sqlmodel import Session, select
 
 from app.core.timeutil import iso_utc
-from app.models import Faction, Group, GroupMember, Match, MatchParticipant, User
+from app.models import Expansion, Faction, Group, GroupMember, Match, MatchParticipant, User
 from app.services import faction_service, group_service, membership_service, user_service
 from app.services.scoring import (
     FACTION_MIN_GAMES,
-    MATCH_POINTS_SQL,
     MIN_GAMES,
-    SMOOTHED_SCORE_SQL,
+    RatedMatch,
+    RatedSeat,
+    Replay,
     leaderboard_sort_key,
-    smoothed_score,
+    replay,
 )
 
-# Базовый блок: одна строка на участие с tie-aware очками.
+
-# Учитываются только ЗАВЕРШЁННЫЕ партии (in_progress без мест в статистику не входят).
+def load_history(session: Session, group_id: int | None = None) -> list[RatedMatch]:
-SCORED_CTE = f"""
+    """Завершённые партии в порядке проигрывания: дата игры, момент завершения, id.
-WITH tie AS (
+
-  SELECT mp.match_id AS match_id, mp.place AS place, COUNT(*) AS tie_size
+    Один запрос на партии с участниками. In_progress в рейтинг не входят: мест у них нет."""
-  FROM match_participants mp
+    stmt = (
-  JOIN matches m ON m.id = mp.match_id
+        select(Match, MatchParticipant)
-  WHERE m.status = 'finished' AND mp.place IS NOT NULL
+        .join(MatchParticipant, MatchParticipant.match_id == Match.id)
-  GROUP BY mp.match_id, mp.place
+        .where(Match.status == "finished")
-),
+    )
-scored AS (
+    if group_id is not None:
-  SELECT mp.user_id      AS user_id,
+        stmt = stmt.where(Match.group_id == group_id)
-         mp.faction_id   AS faction_id,
+    stmt = stmt.order_by(Match.played_at, Match.finished_at, Match.id, MatchParticipant.id)
-         m.id            AS match_id,
+
-         m.group_id      AS group_id,
+    history: list[RatedMatch] = []
-         m.played_at     AS played_at,
+    current: Match | None = None
-         mp.place        AS place,
+    seats: list[RatedSeat] = []
-         mp.was_random   AS was_random,
+
-         m.player_count  AS player_count,
+    def flush() -> None:
-         ({MATCH_POINTS_SQL}) AS points,
+        if current is not None and seats:
-         CASE WHEN mp.place = 1 THEN 1 ELSE 0 END AS is_win
+            history.append(
-  FROM match_participants mp
+                RatedMatch(
-  JOIN matches m ON m.id = mp.match_id
+                    seats=tuple(seats),
-  JOIN tie t ON t.match_id = mp.match_id AND t.place = mp.place
+                    win_reason=current.win_reason,
-  WHERE m.status = 'finished'
+                    end_round=current.end_round,
-)
+                    nine_rounds_rule=current.nine_rounds_rule,
-"""
+                    id=current.id,  # type: ignore[arg-type]
+                    group_id=current.group_id,
+                    played_at=current.played_at,
+                    finished_at=current.finished_at,
+                )
+            )
+
+    for m, p in session.exec(stmt).all():
+        if current is None or m.id != current.id:
+            flush()
+            current, seats = m, []
+        if p.place is None:  # у завершённой партии мест без значения не бывает
+            continue
+        seats.append(
+            RatedSeat(
+                user_id=p.user_id,
+                place=p.place,
+                faction_id=p.faction_id,
+                eliminated=p.eliminated,
+                objectives=p.objectives,
+                worlds=p.worlds,
+            )
+        )
+    flush()
+    return history
 
 
-def _round(value: Any, ndigits: int) -> float | None:
+def _for_group(history: list[RatedMatch], group_id: int) -> list[RatedMatch]:
-    return None if value is None else round(float(value), ndigits)
+    return [m for m in history if m.group_id == group_id]
 
 
-def _normalize(row: dict) -> dict:
+def _user_seats(history: list[RatedMatch], user_id: int) -> list[tuple[RatedMatch, RatedSeat]]:
-    # avatar_url мирроринг user_service.avatar_url_for: версия = epoch(updated_at) из SQL.
+    return [(m, s) for m in history for s in m.seats if s.user_id == user_id]
-    avatar_url = None
+
-    if row.get("avatar_path"):
+
-        avatar_url = f"/api/users/{row['user_id']}/avatar?v={int(row.get('avatar_version') or 0)}"
+def _summary(seats: list[tuple[RatedMatch, RatedSeat]], rating: float | None) -> dict:
+    """Итог игрока: игры, победы, винрейт, среднее место и рейтинг целым числом.
+
+    В расчёте рейтинг без округления (иначе ошибка копилась бы по цепочке партий),
+    показывается — целым."""
+    games = len(seats)
+    if games == 0:
+        return {"games": 0, "wins": 0, "win_rate": 0.0, "avg_place": None, "score": None}
+    wins = sum(1 for _m, s in seats if s.place == 1)
     return {
-        "user_id": row["user_id"],
+        "games": games,
-        "nickname": row["nickname"],
+        "wins": wins,
-        "games": int(row["games"] or 0),
+        "win_rate": round(wins / games, 4),
-        "wins": int(row["wins"] or 0),
+        "avg_place": round(sum(s.place for _m, s in seats) / games, 2),
-        "win_rate": _round(row["win_rate"] or 0.0, 4),
+        "score": None if rating is None else round(rating),
-        "avg_place": _round(row["avg_place"], 2),
-        "score": _round(row["score"], 1),
-        "avatar_url": avatar_url,
     }
 
 
-def _leaderboard_rows(session: Session, group_id: int | None) -> list[dict]:
+def leaderboard(
-    where = "WHERE s.group_id = :gid" if group_id is not None else ""
+    session: Session,
-    sql = f"""
+    group_id: int | None = None,
-    {SCORED_CTE}
+    *,
-    SELECT u.id AS user_id, u.nickname AS nickname,
+    history: list[RatedMatch] | None = None,
-           u.avatar_path AS avatar_path,
+    rep: Replay | None = None,
-           CAST(strftime('%s', u.updated_at) AS INTEGER) AS avatar_version,
+) -> dict:
-           COUNT(*) AS games,
+    """Топ: общий или группы. history/rep — уже посчитанные (home их переиспользует)."""
-           SUM(s.is_win) AS wins,
+    if history is None:
-           AVG(CAST(s.is_win AS FLOAT)) AS win_rate,
+        history = load_history(session, group_id)
-           AVG(s.place) AS avg_place,
+    if rep is None:
-           {SMOOTHED_SCORE_SQL} AS score
+        rep = replay(history)
-    FROM scored s
-    JOIN users u ON u.id = s.user_id
-    {where}
-    GROUP BY u.id, u.nickname, u.avatar_path, u.updated_at
-    """
-    params = {"gid": group_id} if group_id is not None else {}
-    result = session.execute(text(sql), params).mappings().all()
-    return [_normalize(dict(r)) for r in result]
 
+    by_user: dict[int, list[tuple[RatedMatch, RatedSeat]]] = defaultdict(list)
+    for m in history:
+        for s in m.seats:
+            by_user[s.user_id].append((m, s))
+    users = (
+        {u.id: u for u in session.exec(select(User).where(User.id.in_(list(by_user)))).all()}
+        if by_user
+        else {}
+    )
 
-def leaderboard(session: Session, group_id: int | None = None) -> dict:
+    rows = []
-    rows = _leaderboard_rows(session, group_id)
+    for uid, seats in by_user.items():
-    qualified = [r for r in rows if r["games"] >= MIN_GAMES]
+        u = users[uid]
-    provisional = [r for r in rows if r["games"] < MIN_GAMES]
+        rows.append(
-    qualified.sort(key=leaderboard_sort_key)
+            {
-    provisional.sort(key=leaderboard_sort_key)
+                "user_id": uid,
+                "nickname": u.nickname,
+                **_summary(seats, rep.ratings[uid]),
+                "rating": rep.ratings[uid],  # только для сортировки
+                "avatar_url": user_service.avatar_url_for(u.id, u.avatar_path, u.updated_at),  # type: ignore[arg-type]
+            }
+        )
+    qualified = sorted((r for r in rows if r["games"] >= MIN_GAMES), key=leaderboard_sort_key)
+    provisional = sorted((r for r in rows if r["games"] < MIN_GAMES), key=leaderboard_sort_key)
     for i, r in enumerate(qualified, start=1):
         r["rank"] = i
     for r in provisional:
         r["rank"] = None
+    for r in rows:
+        del r["rating"]
     return {
         "entries": qualified,
         "provisional": provisional,
@@ -107,72 +153,38 @@ def leaderboard(session: Session, group_id: int | None = None) -> dict:
     }
 
 
-def _user_scored_rows(session: Session, user_id: int) -> list[dict]:
+def _faction_breakdown(
-    """Строки участия игрока со всеми полями, нужными витринам профиля.
+    session: Session, seats: list[tuple[RatedMatch, RatedSeat]], rep: Replay, user_id: int
-
+) -> list[dict]:
-    Один проход по SCORED_CTE вместо трёх: общий итог, разбивка по фракциям и форма
+    by_faction: dict[int, list[tuple[RatedMatch, RatedSeat]]] = defaultdict(list)
-    последних партий считаются из одного и того же набора строк. CTE джойнит участия
+    for m, s in seats:
-    со всеми партиями приложения, поэтому каждый лишний проход дорожает вместе с
+        by_faction[s.faction_id].append((m, s))
-    общим числом партий, а не с числом партий игрока."""
+    if not by_faction:
-    sql = f"""
+        return []
-    {SCORED_CTE}
+    meta = {
-    SELECT s.group_id AS group_id, s.match_id AS match_id, s.played_at AS played_at,
+        f.id: (f, code)
-           s.place AS place, s.player_count AS player_count,
+        for f, code in session.exec(
-           s.points AS points, s.is_win AS is_win,
+            select(Faction, Expansion.code)
-           f.id AS faction_id, f.code AS code, f.name_ru AS name_ru,
+            .join(Expansion, Expansion.id == Faction.expansion_id)
-           e.code AS expansion_code
+            .where(Faction.id.in_(list(by_faction)))
-    FROM scored s
+        ).all()
-    JOIN factions f ON f.id = s.faction_id
-    JOIN expansions e ON e.id = f.expansion_id
-    WHERE s.user_id = :uid
-    """
-    rows = session.execute(text(sql), {"uid": user_id}).mappings().all()
-    return [dict(r) for r in rows]
-
-
-def _for_group(rows: list[dict], group_id: int | None) -> list[dict]:
-    return rows if group_id is None else [r for r in rows if r["group_id"] == group_id]
-
-
-def _overall_from_rows(rows: list[dict]) -> dict:
-    """Тот же итог, что раньше считал SQL: COUNT/SUM/AVG плюс сглаженный рейтинг."""
-    games = len(rows)
-    if games == 0:
-        return {"games": 0, "wins": 0, "win_rate": 0.0, "avg_place": None, "score": None}
-    wins = sum(int(r["is_win"]) for r in rows)
-    return {
-        "games": games,
-        "wins": wins,
-        "win_rate": _round(wins / games, 4),
-        "avg_place": _round(sum(r["place"] for r in rows) / games, 2),
-        "score": _round(smoothed_score(sum(float(r["points"]) for r in rows), games), 1),
     }
-
-
-def _faction_breakdown_from_rows(rows: list[dict]) -> list[dict]:
-    by_faction: dict[int, list[dict]] = {}
-    for r in rows:
-        by_faction.setdefault(r["faction_id"], []).append(r)
     out = []
     for fid, group in by_faction.items():
-        games = len(group)
+        faction, expansion_code = meta[fid]
-        wins = sum(int(r["is_win"]) for r in group)
+        perf = sum(rep.perf[(m.id, user_id)] for m, _s in group) / len(group)
-        meta = group[0]
         out.append(
             {
                 "faction_id": fid,
-                "code": meta["code"],
+                "code": faction.code,
-                "name_ru": meta["name_ru"],
+                "name_ru": faction.name_ru,
-                "expansion_code": meta["expansion_code"],
+                "expansion_code": expansion_code,
-                "games": games,
+                **{k: v for k, v in _summary(group, None).items() if k != "score"},
-                "wins": wins,
+                # Средний результат относительно ожидания (S − E) × 100: насколько игрок
-                "win_rate": _round(wins / games, 4),
+                # на фракции выступает выше рейтинговых шансов — без привязки к рейтингу.
-                "avg_place": _round(sum(r["place"] for r in group) / games, 2),
+                "score": round(perf * 100, 1),
-                # Фракции: чистое среднее (служебная метрика «лучшая/худшая»,
-                # сглаживание задавило бы её к 50).
-                "score": _round(sum(float(r["points"]) for r in group) / games * 100, 1),
                 "name_ru_prepositional": faction_service.prepositional(
-                    meta["code"], meta["name_ru"]
+                    faction.code, faction.name_ru
                 ),
             }
         )
@@ -180,15 +192,15 @@ def _faction_breakdown_from_rows(rows: list[dict]) -> list[dict]:
     return out
 
 
-def _recent_form_from_rows(rows: list[dict], limit: int = 5) -> list[dict]:
+def _recent_form(seats: list[tuple[RatedMatch, RatedSeat]], limit: int = 5) -> list[dict]:
-    recent = sorted(rows, key=lambda r: (str(r["played_at"]), r["match_id"]), reverse=True)
+    recent = sorted(seats, key=lambda ms: (str(ms[0].played_at), ms[0].id), reverse=True)
     return [
         {
-            "place": r["place"],
+            "place": s.place,
-            "player_count": r["player_count"],
+            "player_count": len(m.seats),
-            "played_at": str(r["played_at"]),
+            "played_at": str(m.played_at),
         }
-        for r in recent[:limit]
+        for m, s in recent[:limit]
     ]
 
 
@@ -208,69 +220,73 @@ def _favorite_faction(session: Session, user_id: int) -> dict | None:
     }
 
 
+def _overall(history: list[RatedMatch], rep: Replay, user_id: int) -> dict:
+    return _summary(_user_seats(history, user_id), rep.ratings.get(user_id))
+
+
 def profile_stats(
     session: Session,
     user_id: int,
     group_id: int | None = None,
     *,
-    rows: list[dict] | None = None,
+    history: list[RatedMatch] | None = None,
+    rep: Replay | None = None,
 ) -> dict:
-    """Витрина профиля. rows — уже вытащенные строки игрока (home() их переиспользует)."""
+    """Витрина профиля. history/rep — общая цепочка, уже посчитанная (home их переиспользует);
-    scoped = _for_group(rows if rows is not None else _user_scored_rows(session, user_id), group_id)
+    с group_id считается групповая цепочка."""
-    overall = _overall_from_rows(scoped)
+    if history is None:
-    factions = _faction_breakdown_from_rows(scoped)
+        history = load_history(session)
+    if group_id is not None:
+        history, rep = _for_group(history, group_id), None
+    if rep is None:
+        rep = replay(history)
+    seats = _user_seats(history, user_id)
+    factions = _faction_breakdown(session, seats, rep, user_id)
     qualified = [f for f in factions if f["games"] >= FACTION_MIN_GAMES]
-    best = max(qualified, key=lambda f: (f["score"] or 0)) if qualified else None
+    best = max(qualified, key=lambda f: f["score"]) if qualified else None
-    worst = min(qualified, key=lambda f: (f["score"] or 0)) if qualified else None
+    worst = min(qualified, key=lambda f: f["score"]) if qualified else None
     # «Чаще всего играет на» — самая игранная фракция по всей истории, включая
     # рандомные раздачи.
     main = max(factions, key=lambda f: f["games"]) if factions else None
     return {
         "user_id": user_id,
-        "overall": overall,
+        "overall": _summary(seats, rep.ratings.get(user_id)),
         "factions": factions,
         "best_faction": best,
         "worst_faction": worst,
         "favorite_faction": _favorite_faction(session, user_id),
         "main_faction": main,
-        "recent_form": _recent_form_from_rows(scoped),
+        "recent_form": _recent_form(seats),
         "min_games": MIN_GAMES,
     }
 
 
 def group_stats(session: Session, group_id: int) -> dict:
-    board = leaderboard(session, group_id=group_id)
+    history = load_history(session, group_id)
-    # Нужны только счётчик и дата последней партии — тянуть строки целиком незачем.
+    board = leaderboard(session, group_id, history=history, rep=replay(history))
-    games_count, last_played = session.exec(
+    last_played = max((m.played_at for m in history), default=None)
-        select(func.count(), func.max(Match.played_at)).where(
-            Match.group_id == group_id, Match.status == "finished"
-        )
-    ).one()
-    last_at = str(last_played) if last_played else None
 
+    games: dict[int, int] = defaultdict(int)
+    wins: dict[int, int] = defaultdict(int)
+    for m in history:
+        for s in m.seats:
+            games[s.faction_id] += 1
+            wins[s.faction_id] += s.place == 1
     available_ids = group_service.available_faction_ids(session, group_id)
-    faction_meta = []
+    factions = sorted(
-    sql = f"""
+        session.exec(select(Faction)).all(), key=lambda f: (-games[f.id], f.sort_order)  # type: ignore[index]
-    {SCORED_CTE}
+    )
-    SELECT f.id AS faction_id, f.code AS code, f.name_ru AS name_ru,
+    faction_meta = [
-           COUNT(s.user_id) AS games, SUM(s.is_win) AS wins
+        {
-    FROM factions f
+            "faction_id": f.id,
-    LEFT JOIN scored s ON s.faction_id = f.id AND s.group_id = :gid
+            "code": f.code,
-    GROUP BY f.id, f.code, f.name_ru
+            "name_ru": f.name_ru,
-    ORDER BY games DESC, f.sort_order
+            "games": games[f.id],  # type: ignore[index]
-    """
+            "wins": wins[f.id],  # type: ignore[index]
-    rows = session.execute(text(sql), {"gid": group_id}).mappings().all()
+            "available": f.id in available_ids,
-    for r in rows:
+        }
-        faction_meta.append(
+        for f in factions
-            {
+    ]
-                "faction_id": r["faction_id"],
-                "code": r["code"],
-                "name_ru": r["name_ru"],
-                "games": int(r["games"] or 0),
-                "wins": int(r["wins"] or 0),
-                "available": r["faction_id"] in available_ids,
-            }
-        )
 
     # Участники без завершённых партий — отдельным блоком (нули, rank=null).
     played_ids = {e["user_id"] for e in board["entries"]} | {
@@ -296,8 +312,8 @@ def group_stats(session: Session, group_id: int) -> dict:
 
     return {
         "group_id": group_id,
-        "total_matches": games_count,
+        "total_matches": len(history),
-        "last_match_at": last_at,
+        "last_match_at": str(last_played) if last_played else None,
         "leaderboard": board["entries"],
         "provisional": board["provisional"],
         "inactive": inactive,
@@ -392,19 +408,15 @@ def user_match_list(
 ) -> dict:
     """История партий игрока: только ЗАВЕРШЁННЫЕ, свежие сверху.
 
-    best_only — одна лучшая партия по League Points (s.points из SCORED_CTE учитывает
+    best_only — одна лучшая партия: наибольший прирост рейтинга в общей цепочке
-    место и размер стола); при равных очках берём более свежую."""
+    (учитывает и соперников, и ход партии); при равенстве берём более свежую."""
     if best_only:
-        sql = f"""
+        history = load_history(session)
-        {SCORED_CTE}
+        rep = replay(history)
-        SELECT s.match_id AS match_id
+        candidates = [
-        FROM scored s
+            (rep.delta[(m.id, user_id)], m.played_at, m.id) for m, _s in _user_seats(history, user_id)
-        WHERE s.user_id = :uid
+        ]
-        ORDER BY s.points DESC, s.played_at DESC, s.match_id DESC
+        matches = [session.get(Match, max(candidates)[2])] if candidates else []
-        LIMIT 1
-        """
-        row = session.execute(text(sql), {"uid": user_id}).mappings().first()
-        matches = [session.get(Match, row["match_id"])] if row else []
         return {
             "items": _match_items(session, matches),
             "total": len(matches),
@@ -466,18 +478,21 @@ def user_in_progress_matches(session: Session, user_id: int) -> list[dict]:
 
 
 def home(session: Session, user_id: int, active_group_id: int | None, leaderboard_limit: int = 10) -> dict:
-    board = leaderboard(session, group_id=None)
+    # История грузится и общая цепочка проигрывается один раз: из них и топ, и профиль;
-    # Строки игрока тянем один раз: из них считается и профиль, и итог по активной группе.
+    # итог по активной группе — отдельная (групповая) цепочка по срезу той же истории.
-    rows = _user_scored_rows(session, user_id)
+    history = load_history(session)
-    profile = profile_stats(session, user_id, group_id=None, rows=rows)
+    rep = replay(history)
+    board = leaderboard(session, history=history, rep=rep)
+    profile = profile_stats(session, user_id, history=history, rep=rep)
     active_group_brief = None
     if active_group_id is not None:
         group = session.get(Group, active_group_id)
         if group is not None:
+            group_history = _for_group(history, active_group_id)
             active_group_brief = {
                 "id": group.id,
                 "name": group.name,
-                **_overall_from_rows(_for_group(rows, active_group_id)),
+                **_overall(group_history, replay(group_history), user_id),
             }
     return {
         "leaderboard": board["entries"][:leaderboard_limit],

backend/tests/test_profile.py

@@ -322,9 +322,10 @@ def test_history_excludes_matches_without_the_player(client: TestClient, engine)
 
 
 def test_history_best_mode_picks_highest_points(client: TestClient, engine):
-    """Режим best берёт партию с максимальными League Points, а не самую свежую.
+    """Режим best берёт партию с наибольшим приростом рейтинга, а не самую свежую.
 
-    Второе место из четырёх даёт (4-2)/3 ≈ 0.67, второе из двух — (2-2)/1 = 0."""
+    Второе место из четырёх равных приносит рейтинг (обыграны двое), второе место
+    в дуэли — отнимает."""
     me = login(client, "Лучший")
     gid, (a, b, c), fids = _group_with(client, engine, "А", "Б", "В")
 

backend/tests/test_rating_stats.py

@@ -0,0 +1,140 @@
+"""Рейтинг в витринах (#23): Elo по упорядоченной истории, общая и групповая цепочки."""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from app.services.scoring import RatedMatch, RatedSeat, replay
+from tests.conftest import add_group_member, create_finished_match, csrf_headers, login
+
+
+def _group(client: TestClient, name: str = "Группа") -> tuple[int, list[int]]:
+    gid = client.post(
+        "/api/groups", json={"name": name, "expansion_ids": []}, headers=csrf_headers(client)
+    ).json()["id"]
+    fids = [f["id"] for f in client.get(f"/api/groups/{gid}/factions").json()]
+    return gid, fids
+
+
+def _duel(client: TestClient, gid: int, fids: list[int], winner: int, loser: int) -> dict:
+    return create_finished_match(
+        client, gid,
+        [
+            {"user_id": winner, "faction_id": fids[0], "place": 1},
+            {"user_id": loser, "faction_id": fids[1], "place": 2},
+        ],
+    )
+
+
+def _board(client: TestClient, path: str = "/api/stats/leaderboard") -> dict[int, dict]:
+    data = client.get(path).json()
+    rows = data["entries"] + data["provisional"] if "entries" in data else (
+        data["leaderboard"] + data["provisional"]
+    )
+    return {e["user_id"]: e for e in rows}
+
+
+def test_newcomer_duel_moves_rating_by_32(client: TestClient, engine):
+    """Первая дуэль новичков на 1500: K = 64, ожидание 0.5, деталей нет → ±32."""
+    me = login(client, "Хозяин")
+    gid, fids = _group(client)
+    b = add_group_member(engine, gid, "Гость")
+    _duel(client, gid, fids, me["id"], b)
+
+    board = client.get("/api/stats/leaderboard").json()
+    by_id = {e["user_id"]: e for e in board["entries"] + board["provisional"]}
+    assert by_id[me["id"]]["score"] == 1532
+    assert by_id[b]["score"] == 1468
+    # 1 игра < MIN_GAMES=10 → оба пока «Новички», ранжированный топ пуст.
+    assert board["entries"] == []
+    assert board["min_games"] == 10
+
+    prof = client.get("/api/users/me/stats").json()
+    assert prof["overall"]["score"] == 1532
+    # Фракционная метрика — S − E: победа при шансах 0.5 даёт +0.5 → 50.0.
+    assert prof["factions"][0]["score"] == 50.0
+
+
+def test_ranked_after_min_games(client: TestClient, engine):
+    me = login(client, "Чемпион")
+    gid, fids = _group(client)
+    b = add_group_member(engine, gid, "Спарринг")
+    for _ in range(10):
+        _duel(client, gid, fids, me["id"], b)
+    board = client.get("/api/stats/leaderboard").json()
+    ranks = {e["user_id"]: (e["rank"], e["score"]) for e in board["entries"]}
+    assert ranks[me["id"]][0] == 1 and ranks[b][0] == 2
+    assert ranks[me["id"]][1] > 1500 > ranks[b][1]
+    assert isinstance(ranks[me["id"]][1], int)
+
+
+def test_editing_past_match_recalculates_later_ones(client: TestClient, engine):
+    """Рейтинг — функция истории: правка первой партии меняет итог после второй."""
+    me = login(client, "А")
+    gid, fids = _group(client)
+    b = add_group_member(engine, gid, "Б")
+    first = _duel(client, gid, fids, me["id"], b)
+    _duel(client, gid, fids, me["id"], b)
+
+    def expected(first_winner: int, first_loser: int) -> dict[int, int]:
+        rep = replay([
+            RatedMatch((RatedSeat(first_winner, 1), RatedSeat(first_loser, 2)), "objectives", id=1),
+            RatedMatch((RatedSeat(me["id"], 1), RatedSeat(b, 2)), "objectives", id=2),
+        ])
+        return {uid: round(r) for uid, r in rep.ratings.items()}
+
+    before = expected(me["id"], b)
+    assert {uid: e["score"] for uid, e in _board(client).items()} == before
+
+    rows = [
+        {"user_id": b, "faction_id": fids[1], "place": 1},
+        {"user_id": me["id"], "faction_id": fids[0], "place": 2},
+    ]
+    r = client.patch(
+        f"/api/matches/{first['id']}",
+        json={"participants": rows, "win_reason": "objectives"},
+        headers=csrf_headers(client),
+    )
+    assert r.status_code == 200, r.text
+    after = expected(b, me["id"])
+    assert after != before
+    assert {uid: e["score"] for uid, e in _board(client).items()} == after
+
+
+def test_group_rating_is_its_own_chain(client: TestClient, engine):
+    """Групповой рейтинг проигрывает только партии группы, общий — все партии."""
+    me = login(client, "Путешественник")
+    g1, f1 = _group(client, "Первая")
+    g2, f2 = _group(client, "Вторая")
+    b = add_group_member(engine, g1, "Сосед")
+    c = add_group_member(engine, g2, "Соседка")
+    _duel(client, g1, f1, me["id"], b)  # в первой группе — победа
+    _duel(client, g2, f2, c, me["id"])  # во второй — поражение
+
+    assert _board(client, f"/api/groups/{g1}/stats")[me["id"]]["score"] == 1532
+    assert _board(client, f"/api/groups/{g2}/stats")[me["id"]]["score"] == 1468
+    overall = _board(client)[me["id"]]["score"]
+    assert overall not in (1532, 1468)
+    assert overall == round(
+        replay([
+            RatedMatch((RatedSeat(me["id"], 1), RatedSeat(b, 2)), "objectives", id=1),
+            RatedMatch((RatedSeat(c, 1), RatedSeat(me["id"], 2)), "objectives", id=2),
+        ]).ratings[me["id"]]
+    )
+
+
+def test_best_match_is_biggest_rating_gain(client: TestClient, engine):
+    """Лучшая партия — наибольший прирост рейтинга, а не свежая из равных побед.
+
+    Обе партии — победы в дуэли. Первая — новичком над равным (+32), вторая — уже
+    с рейтингом 1532 и меньшим K над новичком (≈ +28): лучше первая."""
+    me = login(client, "Лучший")
+    gid, fids = _group(client)
+    x = add_group_member(engine, gid, "Икс")
+    y = add_group_member(engine, gid, "Игрек")
+    first = _duel(client, gid, fids, me["id"], x)
+    _duel(client, gid, fids, me["id"], y)
+
+    client.patch("/api/users/me/profile", json={"history_mode": "best"}, headers=csrf_headers(client))
+    data = client.get(f"/api/users/{me['id']}/matches").json()
+    assert data["total"] == 1
+    assert data["items"][0]["id"] == first["id"]

backend/tests/test_scoring_smoothing.py

@@ -1,39 +0,0 @@
-"""Сглаживание рейтинга: score = (C·m + сумма очков) / (C + игр) × 100, C=10, m=0.5."""
-from __future__ import annotations
-
-from fastapi.testclient import TestClient
-
-from tests.conftest import add_group_member, create_finished_match, csrf_headers, login
-
-
-def test_leaderboard_and_profile_score_are_smoothed(client: TestClient, engine):
-    me = login(client, "Хозяин")
-    gid = client.post(
-        "/api/groups", json={"name": "Группа", "expansion_ids": []}, headers=csrf_headers(client)
-    ).json()["id"]
-    b = add_group_member(engine, gid, "Гость")
-    factions = client.get(f"/api/groups/{gid}/factions").json()
-    f1, f2 = factions[0]["id"], factions[1]["id"]
-
-    create_finished_match(
-        client,
-        gid,
-        [
-            {"user_id": me["id"], "faction_id": f1, "place": 1},
-            {"user_id": b, "faction_id": f2, "place": 2},
-        ],
-    )
-
-    board = client.get("/api/stats/leaderboard").json()
-    by_id = {e["user_id"]: e for e in board["entries"] + board["provisional"]}
-    # Победитель: 1 очко за партию → (10·0.5 + 1) / (10 + 1) × 100 = 54.5.
-    # Проигравший: 0 очков → (10·0.5 + 0) / 11 × 100 = 45.5.
-    assert by_id[me["id"]]["score"] == 54.5
-    assert by_id[b]["score"] == 45.5
-    # 1 игра < MIN_GAMES=10 → оба пока «Новички», ранжированный топ пуст.
-    assert board["entries"] == []
-    assert board["min_games"] == 10
-
-    # Профиль показывает тот же сглаженный рейтинг, что и топ.
-    prof = client.get("/api/users/me/stats").json()
-    assert prof["overall"]["score"] == 54.5

backend/tests/test_stats_passes.py

@@ -1,9 +1,9 @@
-"""Статистика профиля: цифры сходятся с лидербордом, а главная не гоняет CTE лишний раз."""
+"""Статистика профиля: цифры сходятся с лидербордом, а главная не грузит историю лишний раз."""
 from __future__ import annotations
 
 from fastapi.testclient import TestClient
-from sqlalchemy import event
 
+from app.services import stats_service
 from tests.conftest import add_group_member, create_finished_match, csrf_headers, login
 
 
@@ -31,10 +31,7 @@ def _group_with_matches(client: TestClient, engine, games: int = 3) -> tuple[dic
 
 
 def test_profile_numbers_match_leaderboard(client: TestClient, engine):
-    """Профиль считает в Python, лидерборд — в SQL: цифры обязаны совпадать.
+    """Профиль и лидерборд собирают итог разными путями: цифры обязаны совпадать."""
-
-    Формула сглаженного рейтинга живёт в двух видах (SMOOTHED_SCORE_SQL и
-    scoring.smoothed_score); этот тест ловит их расхождение."""
     me, gid, p2 = _group_with_matches(client, engine, games=4)
 
     profile = client.get("/api/users/me/stats").json()["overall"]
@@ -47,33 +44,25 @@ def test_profile_numbers_match_leaderboard(client: TestClient, engine):
         assert profile[field] == entry[field], field
 
 
-def test_home_does_not_repeat_scored_cte(client: TestClient, engine):
+def test_home_loads_history_once(client: TestClient, engine, monkeypatch):
-    """Главная делает не больше двух проходов по SCORED_CTE.
+    """Главная грузит историю партий один раз: топ, профиль и итог активной группы
-
+    считаются из неё. Без этой проверки лишняя загрузка тихо вернётся при правке витрин."""
-    Было пять: лидерборд, три запроса профиля и итог по активной группе. Без этой
-    проверки оптимизация тихо отъедет назад при следующей правке витрин."""
     me, gid, p2 = _group_with_matches(client, engine, games=2)
     client.put(
         "/api/users/me/active-group", json={"group_id": gid}, headers=csrf_headers(client)
     )
 
-    seen: list[str] = []
+    calls: list[int | None] = []
+    original = stats_service.load_history
 
-    def before_execute(conn, cursor, statement, params, context, executemany):
+    def spy(session, group_id=None):
-        if "WITH tie AS" in statement:
+        calls.append(group_id)
-            seen.append(statement)
+        return original(session, group_id)
 
-    event.listen(engine, "before_cursor_execute", before_execute)
+    monkeypatch.setattr(stats_service, "load_history", spy)
-    try:
+    r = client.get("/api/home")
-        r = client.get("/api/home")
+    assert r.status_code == 200, r.text
-        assert r.status_code == 200, r.text
+    assert calls == [None]
-    finally:
-        event.remove(engine, "before_cursor_execute", before_execute)
-
-    # Сейчас ровно два: лидерборд и один проход по строкам игрока. Нижняя граница не
-    # для красоты — без неё тест пройдёт и когда счётчик молча перестанет что-либо
-    # ловить (сменился путь, переименован CTE).
-    assert 1 <= len(seen) <= 2, f"ожидали 1–2 прохода, получили {len(seen)}"
     # Главная всё ещё показывает и профиль, и блок активной группы.
     body = r.json()
     assert body["profile"]["overall"]["games"] == 2