feat: periods, records and the crown

The queries the statistics are made of, all derived rather than cached -- two sources of truth for one number is how a leaderboard comes to disagree with itself, and this history is small enough that recomputing is free. Periods take an inclusive span of stored Warsaw day strings, never a range of timestamps, so a daylight-saving change cannot move a boundary. Trend reports None rather than 0 when there is no previous period: "nothing to compare against yet" and "no change" are different statements and the group will read them differently. A tie shares the crown and keeps every tied run alive, per kuhy's call. Streaks count *recorded* days, so a day nobody spoke does not end a run -- there was no crown to lose -- and neither does a day the bot was offline for, which it could not have judged. Every record excludes the backfilled day. It is a 22-hour window filed as a day, and leaving it in would hand somebody a personal best they never set. Two bugs found by running the code rather than reading it: - current_streaks reported every streak as zero. Ending somebody's run dropped them from the tally along with the length they had already accumulated, so the surviving dict was empty by the time it returned. Now the run ends and the count it reached is kept. - daily_winners depended on SQLite's grouping order to find a day's leader. It happened to work because uuids sorted conveniently in the fixtures. Ordering by messages DESC makes the first row per day the leader by construction, which is both correct and obviously so. No suppressions: the period queries compare an exclusion parameter against '' instead of pasting half a WHERE clause into an f-string, which is a SQL-injection shape even when every input is a constant. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015d48t4qsQLBJnDwDeo2Nom
2026-11-10 19:27:20 +01:00 · 2026-08-30 19:38:54 +02:00 · 2026-08-30 19:38:54 +02:00 · a9ffd02f33
commit a9ffd02f33
parent 0f824c441e
7 changed files with 921 additions and 0 deletions
--- a/signal_bot/stats/queries_period.py
+++ b/signal_bot/stats/queries_period.py
@ -0,0 +1,173 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""Weeks, months, years, and the per-weekday picture.
+
+Every query takes an inclusive span of stored Warsaw day strings, so a
+daylight-saving change cannot move a boundary -- see ``stats.periods``.
+
+Several of them take ``exclude``: the one day carried over from the pre-0.1.0
+counter has real totals but no hours, no lengths, and a 22-hour window
+pretending to be a day. It belongs in a total and nowhere near an average.
+
+The exclusion is a bound parameter compared against ``''`` rather than a
+clause pasted into the statement. No real day is the empty string, so one
+static query serves both cases -- and the alternative, an f-string holding
+half a WHERE clause, is a SQL-injection shape even when every input is a
+module constant.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from signal_bot.stats.periods import weekday_of
+from signal_bot.stats.queries_day import DayRow
+
+if TYPE_CHECKING:
+    from signal_bot.stats.events import EventStore
+    from signal_bot.stats.periods import Period
+
+PERIOD_ROWS = """
+    SELECT uuid,
+           display_name,
+           COUNT(*)         AS messages,
+           SUM(media_count) AS media,
+           SUM(is_command)  AS commands,
+           SUM(char_len)    AS chars
+    FROM message_events
+    WHERE group_id = ? AND day BETWEEN ? AND ? AND day <> ?
+    GROUP BY uuid
+    ORDER BY messages DESC, display_name ASC
+"""
+
+DAILY_SERIES = """
+    SELECT day, COUNT(*) AS messages
+    FROM message_events
+    WHERE group_id = ? AND day BETWEEN ? AND ? AND day <> ?
+    GROUP BY day
+    ORDER BY day
+"""
+
+EVERY_DAY = """
+    SELECT day, COUNT(*) AS messages
+    FROM message_events
+    WHERE group_id = ? AND day <> ?
+    GROUP BY day
+    ORDER BY day
+"""
+
+
+def period_rows(
+    events: EventStore,
+    group_id: str,
+    period: Period,
+    *,
+    exclude: str | None = None,
+) -> list[DayRow]:
+    """Everyone who wrote during a period, most talkative first."""
+    rows = events.connection.execute(
+        PERIOD_ROWS, (group_id, period.start, period.end, exclude or "")
+    ).fetchall()
+    return [
+        DayRow(
+            uuid=r["uuid"],
+            display_name=r["display_name"],
+            messages=r["messages"],
+            media=r["media"],
+            commands=r["commands"],
+            chars=r["chars"],
+        )
+        for r in rows
+    ]
+
+
+def daily_series(
+    events: EventStore,
+    group_id: str,
+    start: str,
+    end: str,
+    *,
+    exclude: str | None = None,
+) -> dict[str, int]:
+    """Messages per day across a span, missing days absent rather than zero.
+
+    The caller decides what a missing day means: for a streak it is silence,
+    for an average it is a day nobody was there to record.
+    """
+    rows = events.connection.execute(DAILY_SERIES, (group_id, start, end, exclude or "")).fetchall()
+    return {r["day"]: r["messages"] for r in rows}
+
+
+def every_day(
+    events: EventStore,
+    group_id: str,
+    *,
+    exclude: str | None = None,
+) -> dict[str, int]:
+    """Messages per day across the whole history."""
+    rows = events.connection.execute(EVERY_DAY, (group_id, exclude or "")).fetchall()
+    return {r["day"]: r["messages"] for r in rows}
+
+
+def weekday_buckets(days: dict[str, int]) -> dict[int, list[int]]:
+    """Bucket a day series by weekday, 0=Monday through 6=Sunday.
+
+    Returns the days themselves rather than a mean, so a caller can see how
+    many a figure rests on: "wtorki mają średnio 74" off a single Tuesday is a
+    number that should not be said out loud.
+    """
+    buckets: dict[int, list[int]] = {}
+    for day, messages in days.items():
+        buckets.setdefault(weekday_of(day), []).append(messages)
+    return buckets
+
+
+def total_for(
+    events: EventStore,
+    group_id: str,
+    start: str,
+    end: str,
+    *,
+    exclude: str | None = None,
+) -> int:
+    """How many messages a span holds."""
+    return sum(daily_series(events, group_id, start, end, exclude=exclude).values())
+
+
+@dataclass(frozen=True, slots=True)
+class Trend:
+    """One period measured against the one before it."""
+
+    now: int
+    before: int
+
+    @property
+    def delta(self) -> int:
+        """Absolute change, which may be negative."""
+        return self.now - self.before
+
+    @property
+    def percent(self) -> int | None:
+        """Rounded percentage change, or None when there is no baseline.
+
+        None rather than 0: "nothing to compare against yet" is a different
+        statement from "no change", and the renderers say each differently.
+        """
+        if not self.before:
+            return None
+        return round(self.delta / self.before * 100)
+
+    @property
+    def arrow(self) -> str:
+        """An arrow, or "" when there is nothing to compare against."""
+        if self.percent is None or self.delta == 0:
+            return ""
+        return "↑" if self.delta > 0 else "↓"
+
+
+def share_of_voice(rows: list[DayRow]) -> list[tuple[str, int]]:
+    """Each person's percentage of the messages, largest first."""
+    total = sum(row.messages for row in rows)
+    if not total:
+        return []
+    return [(row.display_name, round(row.messages / total * 100)) for row in rows]
--- a/signal_bot/stats/records.py
+++ b/signal_bot/stats/records.py
@ -0,0 +1,184 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""All-time records: busiest days, quietest days, personal bests, silences.
+
+Everything here is derived by query rather than kept in a cache table. Two
+sources of truth for one number is how a leaderboard comes to disagree with
+itself, and the history is small enough that recomputing is free.
+
+The backfilled day is excluded from every record. It is a 22-hour window
+filed as a day, so leaving it in would hand somebody a personal best they
+never set -- see ``stats.backfill``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from signal_bot.stats.periods import shift
+
+if TYPE_CHECKING:
+    from signal_bot.stats.events import EventStore
+
+PERSON_DAYS = """
+    SELECT day, COUNT(*) AS messages
+    FROM message_events
+    WHERE group_id = ? AND uuid = ? AND day <> ?
+    GROUP BY day
+    ORDER BY messages DESC, day ASC
+"""
+
+MEDIA_DAYS = """
+    SELECT day, SUM(media_count) AS media
+    FROM message_events
+    WHERE group_id = ? AND day <> ?
+    GROUP BY day
+    HAVING media > 0
+    ORDER BY media DESC, day ASC
+"""
+
+DAILY_LEADERS = """
+    SELECT day, uuid, display_name, COUNT(*) AS messages
+    FROM message_events
+    WHERE group_id = ? AND day <> ?
+    GROUP BY day, uuid
+    ORDER BY day ASC
+"""
+
+
+@dataclass(frozen=True, slots=True)
+class DayRecord:
+    """One day and the number that makes it a record."""
+
+    day: str
+    value: int
+
+
+def busiest_day(days: dict[str, int]) -> DayRecord | None:
+    """The day with the most messages, earliest day winning a tie."""
+    if not days:
+        return None
+    day = min(days, key=lambda d: (-days[d], d))
+    return DayRecord(day, days[day])
+
+
+def quietest_day(days: dict[str, int]) -> DayRecord | None:
+    """The quietest day on which anybody spoke at all.
+
+    Days with no messages are absent from the series rather than zero, and
+    that is the right reading: a day nobody spoke is a silence, which
+    ``longest_silence`` measures. A "quietest day" of zero would just name
+    whichever day the bot happened to be offline.
+    """
+    if not days:
+        return None
+    day = min(days, key=lambda d: (days[d], d))
+    return DayRecord(day, days[day])
+
+
+def personal_best(
+    events: EventStore,
+    group_id: str,
+    uuid: str,
+    *,
+    exclude: str | None = None,
+) -> DayRecord | None:
+    """Somebody's busiest day ever, or None if they have never written."""
+    row = events.connection.execute(PERSON_DAYS, (group_id, uuid, exclude or "")).fetchone()
+    return DayRecord(row["day"], row["messages"]) if row else None
+
+
+def previous_best(
+    events: EventStore,
+    group_id: str,
+    uuid: str,
+    before: str,
+    *,
+    exclude: str | None = None,
+) -> DayRecord | None:
+    """Somebody's best day *excluding* one day, for "poprzedni rekord".
+
+    Announcing "rekord życiowy: 76" is worth little without what it beat, and
+    the day being announced must not be its own previous best.
+    """
+    rows = events.connection.execute(PERSON_DAYS, (group_id, uuid, exclude or "")).fetchall()
+    for row in rows:
+        if row["day"] != before:
+            return DayRecord(row["day"], row["messages"])
+    return None
+
+
+def most_media_day(
+    events: EventStore,
+    group_id: str,
+    *,
+    exclude: str | None = None,
+) -> DayRecord | None:
+    """The day carrying the most photos and videos."""
+    row = events.connection.execute(MEDIA_DAYS, (group_id, exclude or "")).fetchone()
+    return DayRecord(row["day"], row["media"]) if row else None
+
+
+@dataclass(frozen=True, slots=True)
+class Silence:
+    """A stretch of days on which nobody wrote anything."""
+
+    start: str
+    end: str
+    days: int
+
+
+def longest_silence(days: dict[str, int], until: str) -> Silence | None:
+    """The longest run of consecutive silent days, up to and including ``until``.
+
+    Walks the calendar rather than the recorded days, because the whole point
+    is the days that are *missing* from the series. Returns None when the
+    history is too short to contain a gap.
+    """
+    if not days:
+        return None
+    spoke = set(days)
+    first = min(spoke)
+    best: Silence | None = None
+    run_start: str | None = None
+    day = first
+
+    while day <= until:
+        if day in spoke:
+            if run_start is not None:
+                run = _silence(run_start, shift(day, -1))
+                best = run if best is None or run.days > best.days else best
+                run_start = None
+        elif run_start is None:
+            run_start = day
+        day = shift(day, 1)
+
+    if run_start is not None:
+        run = _silence(run_start, until)
+        best = run if best is None or run.days > best.days else best
+    return best
+
+
+def _silence(start: str, end: str) -> Silence:
+    """Build a Silence, counting both ends."""
+    length = 1
+    day = start
+    while day < end:
+        day = shift(day, 1)
+        length += 1
+    return Silence(start, end, length)
+
+
+def silent_since(days: dict[str, int], uuid_days: dict[str, int], until: str) -> int:
+    """How many days somebody has gone without writing, as of ``until``.
+
+    Counts back from ``until`` to their last message. Zero means they wrote
+    today; the group's own recorded days bound it, so a person who joined
+    yesterday is not reported as silent for a year.
+    """
+    if not days:
+        return 0
+    if not uuid_days:
+        return sum(1 for day in days if day <= until)
+    last = max(uuid_days)
+    return sum(1 for day in days if last < day <= until)
--- a/signal_bot/stats/streaks.py
+++ b/signal_bot/stats/streaks.py
@ -0,0 +1,126 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""Who holds the crown, and for how long.
+
+A tie shares the crown and keeps every tied streak alive. That is a decision,
+not an oversight: breaking somebody's forty-day run on a technicality is the
+kind of thing that starts an argument in a group chat, and "we were both first"
+is what the people involved would say anyway.
+
+Streaks count *recorded* days, not calendar days. A day nobody spoke does not
+end a run -- there was no crown to lose -- and neither does a day the bot was
+offline for, which it could not have judged.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from signal_bot.stats.events import EventStore
+
+DAILY_COUNTS = """
+    SELECT day, uuid, display_name, COUNT(*) AS messages
+    FROM message_events
+    WHERE group_id = ? AND day <> ?
+    GROUP BY day, uuid
+    ORDER BY day ASC, messages DESC, display_name ASC
+"""
+
+
+@dataclass(frozen=True, slots=True)
+class Streak:
+    """A run of consecutive recorded days holding the crown."""
+
+    uuid: str
+    display_name: str
+    length: int
+    last_day: str
+
+
+def daily_winners(
+    events: EventStore,
+    group_id: str,
+    *,
+    exclude: str | None = None,
+) -> dict[str, list[tuple[str, str]]]:
+    """Who led each recorded day, as (uuid, display_name), ties included."""
+    rows = events.connection.execute(DAILY_COUNTS, (group_id, exclude or "")).fetchall()
+    best: dict[str, int] = {}
+    winners: dict[str, list[tuple[str, str]]] = {}
+    for row in rows:
+        day, count = row["day"], row["messages"]
+        # Ordered messages-first, so the day's leader is always the row that
+        # creates the entry and everything after it either ties or loses.
+        if day not in best:
+            best[day] = count
+            winners[day] = [(row["uuid"], row["display_name"])]
+        elif count == best[day]:
+            winners[day].append((row["uuid"], row["display_name"]))
+    return winners
+
+
+def current_streaks(winners: dict[str, list[tuple[str, str]]]) -> list[Streak]:
+    """Everyone whose run reaches the most recent recorded day.
+
+    Sorted longest first, then by name, so two people on the same run are
+    reported in a stable order rather than whichever the dict happened to
+    yield.
+    """
+    if not winners:
+        return []
+    days = sorted(winners)
+    latest = days[-1]
+    holders = dict(winners[latest])
+
+    lengths = dict.fromkeys(holders, 0)
+    running = set(holders)
+    for day in reversed(days):
+        today = {uuid for uuid, _ in winners[day]}
+        for uuid in sorted(running):
+            if uuid in today:
+                lengths[uuid] += 1
+            else:
+                # Their run ended on this day. What they had already
+                # accumulated is kept -- dropping them from the tally, as an
+                # earlier spelling did, reported every streak as zero.
+                running.discard(uuid)
+        if not running:
+            break
+
+    return sorted(
+        (Streak(uuid, holders[uuid], length, latest) for uuid, length in lengths.items()),
+        key=lambda s: (-s.length, s.display_name),
+    )
+
+
+def longest_ever(winners: dict[str, list[tuple[str, str]]]) -> Streak | None:
+    """The longest crown run anybody has ever put together."""
+    running: dict[str, int] = {}
+    names: dict[str, str] = {}
+    best: Streak | None = None
+
+    for day in sorted(winners):
+        today = dict(winners[day])
+        names.update(today)
+        running = {uuid: running.get(uuid, 0) + 1 for uuid in today}
+        for uuid, length in running.items():
+            if (
+                best is None
+                or length > best.length
+                or (length == best.length and names[uuid] < best.display_name)
+            ):
+                best = Streak(uuid, names[uuid], length, day)
+    return best
+
+
+def held_days(winners: dict[str, list[tuple[str, str]]], uuid: str) -> int:
+    """How many days somebody has held the crown in total, ties included."""
+    return sum(1 for holders in winners.values() if any(u == uuid for u, _ in holders))
+
+
+def names_of(holders: Iterable[tuple[str, str]]) -> list[str]:
+    """Display names of a day's crown holders, alphabetical for stability."""
+    return sorted(name for _, name in holders)
--- a/tests/stats_seed.py
+++ b/tests/stats_seed.py
@ -0,0 +1,64 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""A deterministic history to run the period, record and streak queries over.
+
+Not a conftest fixture: it is a builder several test modules call with a
+different history each, rather than one shared shape. Counts come from
+arithmetic, never from ``random`` -- a fixture that changes between runs turns
+a real failure into a flake.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from signal_bot.stats.events import MessageEvent
+
+if TYPE_CHECKING:
+    from signal_bot.store import Store
+
+GROUP = "group-a=="
+MITHRA = ("u-ala", "Ala")
+KUHY = ("u-kuhy", "kuhy")
+IGA = ("u-iga", "Basia")
+
+
+@dataclass(frozen=True, slots=True)
+class Shape:
+    """The optional detail of a seeded day, so ``wrote`` stays legible."""
+
+    media: int = 0
+    hour: int = 12
+    chars: int = 10
+    commands: int = 0
+    group_id: str = GROUP
+
+
+def wrote(
+    store: Store,
+    day: str,
+    who: tuple[str, str],
+    count: int,
+    shape: Shape | None = None,
+) -> None:
+    """Put ``count`` messages from one person into one group on one day."""
+    shape = shape or Shape()
+    uuid, name = who
+    for index in range(count):
+        store.events.record(
+            MessageEvent(
+                group_id=shape.group_id,
+                uuid=uuid,
+                display_name=name,
+                # Unique within (group, uuid) so the primary key does not
+                # absorb them, and stable so two runs agree.
+                sent_at_ms=int(day.replace("-", "")) * 1000 + index,
+                day=day,
+                local_hour=shape.hour,
+                media_kind="image" if index < shape.media else "",
+                media_count=1 if index < shape.media else 0,
+                char_len=shape.chars,
+                is_command=index < shape.commands,
+            )
+        )
+    store.events.note_person(shape.group_id, uuid, name, "", day)
--- a/tests/test_stats_queries_period.py
+++ b/tests/test_stats_queries_period.py
@ -0,0 +1,121 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""Weeks, months, years, trends and the per-weekday picture."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from signal_bot.stats import periods
+from signal_bot.stats.queries_period import (
+    Trend,
+    daily_series,
+    every_day,
+    period_rows,
+    share_of_voice,
+    total_for,
+    weekday_buckets,
+)
+from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote
+
+if TYPE_CHECKING:
+    from signal_bot.store import Store
+
+# 2026-08-31 is a Monday, so this spans one ISO week exactly.
+WEEK = periods.week_of("2026-09-01")
+PREVIOUS = periods.previous(WEEK)
+
+
+def history(store: Store) -> None:
+    """A full week, plus one day in the week before it."""
+    wrote(store, "2026-08-25", KUHY, 8)
+    wrote(store, "2026-08-31", MITHRA, 10, Shape(media=2, commands=1))
+    wrote(store, "2026-09-01", MITHRA, 20)
+    wrote(store, "2026-09-01", KUHY, 10)
+    wrote(store, "2026-09-02", IGA, 5)
+
+
+def test_period_rows_are_ordered_by_volume(store: Store) -> None:
+    history(store)
+    rows = period_rows(store.events, GROUP, WEEK)
+    assert [(r.display_name, r.messages) for r in rows] == [
+        ("Ala", 30),
+        ("kuhy", 10),
+        ("Basia", 5),
+    ]
+
+
+def test_period_rows_carry_media_commands_and_length(store: Store) -> None:
+    history(store)
+    ala = period_rows(store.events, GROUP, WEEK)[0]
+    assert (ala.media, ala.commands, ala.chars) == (2, 1, 300)
+
+
+def test_a_period_stops_at_its_own_boundary(store: Store) -> None:
+    """The 25th belongs to the previous week and must not leak in."""
+    history(store)
+    assert total_for(store.events, GROUP, WEEK.start, WEEK.end) == 45
+    assert total_for(store.events, GROUP, PREVIOUS.start, PREVIOUS.end) == 8
+
+
+def test_daily_series_omits_days_nobody_spoke(store: Store) -> None:
+    """Absent, not zero: the caller decides what a missing day means."""
+    history(store)
+    series = daily_series(store.events, GROUP, WEEK.start, WEEK.end)
+    assert series == {"2026-08-31": 10, "2026-09-01": 30, "2026-09-02": 5}
+
+
+def test_every_day_spans_the_whole_history(store: Store) -> None:
+    history(store)
+    assert set(every_day(store.events, GROUP)) == {
+        "2026-08-25",
+        "2026-08-31",
+        "2026-09-01",
+        "2026-09-02",
+    }
+
+
+def test_weekday_buckets_index_monday_as_zero(store: Store) -> None:
+    history(store)
+    buckets = weekday_buckets(every_day(store.events, GROUP))
+    # 2026-08-25 is a Tuesday, 08-31 a Monday, 09-01 a Tuesday, 09-02 a Wednesday.
+    assert buckets[0] == [10]
+    assert sorted(buckets[1]) == [8, 30]
+    assert buckets[2] == [5]
+
+
+def test_trend_reports_direction_and_size() -> None:
+    trend = Trend(now=45, before=30)
+    assert (trend.delta, trend.percent, trend.arrow) == (15, 50, "↑")
+    down = Trend(now=15, before=30)
+    assert (down.delta, down.percent, down.arrow) == (-15, -50, "↓")
+
+
+def test_a_flat_period_gets_no_arrow() -> None:
+    assert Trend(now=30, before=30).arrow == ""
+    assert Trend(now=30, before=30).percent == 0
+
+
+def test_no_baseline_is_not_the_same_as_no_change() -> None:
+    """The first week ever has nothing to compare against, and says so."""
+    first = Trend(now=45, before=0)
+    assert first.percent is None
+    assert first.arrow == ""
+    assert first.delta == 45
+
+
+def test_share_of_voice(store: Store) -> None:
+    history(store)
+    rows = period_rows(store.events, GROUP, WEEK)
+    assert share_of_voice(rows) == [("Ala", 67), ("kuhy", 22), ("Basia", 11)]
+
+
+def test_share_of_voice_of_nothing_is_empty() -> None:
+    assert share_of_voice([]) == []
+
+
+def test_the_backfilled_day_can_be_left_out_of_every_query(store: Store) -> None:
+    history(store)
+    assert total_for(store.events, GROUP, WEEK.start, WEEK.end, exclude="2026-09-01") == 15
+    assert "2026-09-01" not in every_day(store.events, GROUP, exclude="2026-09-01")
+    rows = period_rows(store.events, GROUP, WEEK, exclude="2026-09-01")
+    assert [r.display_name for r in rows] == ["Ala", "Basia"]
--- a/tests/test_stats_records.py
+++ b/tests/test_stats_records.py
@ -0,0 +1,135 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""All-time records, and the silences between them."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from signal_bot.stats.queries_period import every_day
+from signal_bot.stats.records import (
+    busiest_day,
+    longest_silence,
+    most_media_day,
+    personal_best,
+    previous_best,
+    quietest_day,
+    silent_since,
+)
+from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, Shape, wrote
+
+if TYPE_CHECKING:
+    from signal_bot.store import Store
+
+
+def history(store: Store) -> None:
+    """Four days, a two-day gap, ties, and somebody who stops writing."""
+    wrote(store, "2026-09-01", MITHRA, 10, Shape(media=3))
+    wrote(store, "2026-09-01", KUHY, 5)
+    wrote(store, "2026-09-02", MITHRA, 4)
+    wrote(store, "2026-09-02", KUHY, 4)
+    wrote(store, "2026-09-03", IGA, 2)
+    # 09-04 and 09-05 silent.
+    wrote(store, "2026-09-06", KUHY, 9, Shape(media=1))
+
+
+def test_no_history_yields_no_records(store: Store) -> None:
+    """Day one of this release: everything empty, nothing crashing."""
+    days = every_day(store.events, GROUP)
+    assert busiest_day(days) is None
+    assert quietest_day(days) is None
+    assert longest_silence(days, "2026-09-06") is None
+    assert most_media_day(store.events, GROUP) is None
+    assert personal_best(store.events, GROUP, "u-kuhy") is None
+
+
+def test_busiest_and_quietest(store: Store) -> None:
+    history(store)
+    days = every_day(store.events, GROUP)
+    assert busiest_day(days).day == "2026-09-01"
+    assert busiest_day(days).value == 15
+    assert quietest_day(days).day == "2026-09-03"
+    assert quietest_day(days).value == 2
+
+
+def test_the_quietest_day_is_one_somebody_spoke_on(store: Store) -> None:
+    """A silent day is a silence, not a quiet day, and is measured separately."""
+    history(store)
+    days = every_day(store.events, GROUP)
+    assert quietest_day(days).value > 0
+    assert "2026-09-04" not in days
+
+
+def test_a_tie_on_the_busiest_day_goes_to_the_earlier(store: Store) -> None:
+    wrote(store, "2026-09-02", KUHY, 7)
+    wrote(store, "2026-09-01", MITHRA, 7)
+    assert busiest_day(every_day(store.events, GROUP)).day == "2026-09-01"
+
+
+def test_personal_best_and_what_it_beat(store: Store) -> None:
+    history(store)
+    best = personal_best(store.events, GROUP, "u-kuhy")
+    assert (best.day, best.value) == ("2026-09-06", 9)
+    previous = previous_best(store.events, GROUP, "u-kuhy", "2026-09-06")
+    assert (previous.day, previous.value) == ("2026-09-01", 5)
+
+
+def test_a_first_ever_day_has_nothing_to_beat(store: Store) -> None:
+    """The "poprzedni rekord" clause has to be omitted, not printed as 0."""
+    wrote(store, "2026-09-01", IGA, 3)
+    assert previous_best(store.events, GROUP, "u-iga", "2026-09-01") is None
+
+
+def test_most_media_day(store: Store) -> None:
+    history(store)
+    record = most_media_day(store.events, GROUP)
+    assert (record.day, record.value) == ("2026-09-01", 3)
+
+
+def test_a_history_with_no_media_has_no_media_record(store: Store) -> None:
+    wrote(store, "2026-09-01", KUHY, 4)
+    assert most_media_day(store.events, GROUP) is None
+
+
+def test_longest_silence(store: Store) -> None:
+    history(store)
+    gap = longest_silence(every_day(store.events, GROUP), "2026-09-06")
+    assert (gap.start, gap.end, gap.days) == ("2026-09-04", "2026-09-05", 2)
+
+
+def test_a_silence_running_to_today_counts(store: Store) -> None:
+    """The gap the group is currently in is the one worth mentioning."""
+    wrote(store, "2026-09-01", KUHY, 1)
+    gap = longest_silence(every_day(store.events, GROUP), "2026-09-05")
+    assert (gap.start, gap.end, gap.days) == ("2026-09-02", "2026-09-05", 4)
+
+
+def test_an_unbroken_history_has_no_silence(store: Store) -> None:
+    wrote(store, "2026-09-01", KUHY, 1)
+    wrote(store, "2026-09-02", KUHY, 1)
+    assert longest_silence(every_day(store.events, GROUP), "2026-09-02") is None
+
+
+def test_silent_since_counts_recorded_days_only(store: Store) -> None:
+    """Bounded by the group's own history, so a newcomer is not "silent for a year"."""
+    history(store)
+    days = every_day(store.events, GROUP)
+    ala = {d: n for d, n in days.items() if d in {"2026-09-01", "2026-09-02"}}
+    assert silent_since(days, ala, "2026-09-06") == 2
+    assert silent_since(days, {"2026-09-06": 9}, "2026-09-06") == 0
+
+
+def test_somebody_who_never_wrote_is_silent_for_the_whole_history(store: Store) -> None:
+    history(store)
+    days = every_day(store.events, GROUP)
+    assert silent_since(days, {}, "2026-09-06") == len(days)
+    assert silent_since({}, {}, "2026-09-06") == 0
+
+
+def test_the_backfilled_day_is_kept_out_of_records(store: Store) -> None:
+    """A 22-hour window filed as a day must not become somebody's best."""
+    wrote(store, "2026-08-30", MITHRA, 98)
+    wrote(store, "2026-09-01", MITHRA, 10)
+    assert personal_best(store.events, GROUP, "u-ala").value == 98
+    excluded = personal_best(store.events, GROUP, "u-ala", exclude="2026-08-30")
+    assert (excluded.day, excluded.value) == ("2026-09-01", 10)
+    assert "2026-08-30" not in every_day(store.events, GROUP, exclude="2026-08-30")
--- a/tests/test_stats_streaks.py
+++ b/tests/test_stats_streaks.py
@ -0,0 +1,118 @@
+# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
+"""The crown: who holds it, for how long, and what a tie does to a run."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from signal_bot.stats.streaks import (
+    current_streaks,
+    daily_winners,
+    held_days,
+    longest_ever,
+    names_of,
+)
+from tests.stats_seed import GROUP, IGA, KUHY, MITHRA, wrote
+
+if TYPE_CHECKING:
+    from signal_bot.store import Store
+
+
+def test_no_history_has_no_crown(store: Store) -> None:
+    winners = daily_winners(store.events, GROUP)
+    assert winners == {}
+    assert current_streaks(winners) == []
+    assert longest_ever(winners) is None
+    assert held_days(winners, "u-kuhy") == 0
+
+
+def test_the_busiest_person_holds_the_day(store: Store) -> None:
+    wrote(store, "2026-09-01", MITHRA, 10)
+    wrote(store, "2026-09-01", KUHY, 5)
+    winners = daily_winners(store.events, GROUP)
+    assert names_of(winners["2026-09-01"]) == ["Ala"]
+
+
+def test_a_tie_shares_the_crown(store: Store) -> None:
+    """kuhy's call: shared, because the alternative starts arguments."""
+    wrote(store, "2026-09-01", MITHRA, 4)
+    wrote(store, "2026-09-01", KUHY, 4)
+    wrote(store, "2026-09-01", IGA, 1)
+    winners = daily_winners(store.events, GROUP)
+    assert names_of(winners["2026-09-01"]) == ["Ala", "kuhy"]
+
+
+def test_a_tie_keeps_every_tied_run_alive(store: Store) -> None:
+    """Three days, both tied throughout: both hold a three-day streak."""
+    for day in ("2026-09-01", "2026-09-02", "2026-09-03"):
+        wrote(store, day, MITHRA, 3)
+        wrote(store, day, KUHY, 3)
+    streaks = current_streaks(daily_winners(store.events, GROUP))
+    assert [(s.display_name, s.length) for s in streaks] == [("Ala", 3), ("kuhy", 3)]
+
+
+def test_a_run_survives_a_day_nobody_spoke(store: Store) -> None:
+    """There was no crown to lose on a silent day, so the run continues."""
+    wrote(store, "2026-09-01", KUHY, 5)
+    # 09-02 and 09-03 silent.
+    wrote(store, "2026-09-04", KUHY, 5)
+    streaks = current_streaks(daily_winners(store.events, GROUP))
+    assert [(s.display_name, s.length) for s in streaks] == [("kuhy", 2)]
+
+
+def test_losing_a_day_ends_a_run_but_keeps_what_it_reached(store: Store) -> None:
+    """The bug this test exists for reported every streak as zero."""
+    wrote(store, "2026-09-01", MITHRA, 10)
+    wrote(store, "2026-09-02", MITHRA, 10)
+    wrote(store, "2026-09-03", MITHRA, 4)
+    wrote(store, "2026-09-03", KUHY, 4)
+    wrote(store, "2026-09-06", KUHY, 9)
+    streaks = current_streaks(daily_winners(store.events, GROUP))
+    assert [(s.display_name, s.length) for s in streaks] == [("kuhy", 2)]
+    assert streaks[0].last_day == "2026-09-06"
+
+
+def test_longest_ever_survives_the_run_that_set_it_ending(store: Store) -> None:
+    wrote(store, "2026-09-01", MITHRA, 10)
+    wrote(store, "2026-09-02", MITHRA, 10)
+    wrote(store, "2026-09-03", MITHRA, 10)
+    wrote(store, "2026-09-04", KUHY, 10)
+    best = longest_ever(daily_winners(store.events, GROUP))
+    assert (best.display_name, best.length, best.last_day) == ("Ala", 3, "2026-09-03")
+
+
+def test_held_days_counts_shared_days_for_everyone_tied(store: Store) -> None:
+    wrote(store, "2026-09-01", MITHRA, 10)
+    wrote(store, "2026-09-02", MITHRA, 4)
+    wrote(store, "2026-09-02", KUHY, 4)
+    winners = daily_winners(store.events, GROUP)
+    assert held_days(winners, "u-ala") == 2
+    assert held_days(winners, "u-kuhy") == 1
+    assert held_days(winners, "u-iga") == 0
+
+
+def test_somebody_who_merely_took_part_holds_nothing(store: Store) -> None:
+    """Neither leader nor tied: the common case, and the one worth pinning."""
+    wrote(store, "2026-09-01", MITHRA, 10)
+    wrote(store, "2026-09-01", KUHY, 4)
+    wrote(store, "2026-09-01", IGA, 1)
+    winners = daily_winners(store.events, GROUP)
+    assert names_of(winners["2026-09-01"]) == ["Ala"]
+    assert held_days(winners, "u-iga") == 0
+
+
+def test_streak_order_is_stable(store: Store) -> None:
+    """Longest first, then by name -- not whichever the dict happened to yield."""
+    wrote(store, "2026-09-01", MITHRA, 3)
+    wrote(store, "2026-09-01", KUHY, 3)
+    wrote(store, "2026-09-01", IGA, 3)
+    streaks = current_streaks(daily_winners(store.events, GROUP))
+    assert [s.display_name for s in streaks] == ["Basia", "Ala", "kuhy"]
+
+
+def test_the_backfilled_day_can_be_left_out(store: Store) -> None:
+    wrote(store, "2026-08-30", MITHRA, 98)
+    wrote(store, "2026-09-01", KUHY, 5)
+    winners = daily_winners(store.events, GROUP, exclude="2026-08-30")
+    assert "2026-08-30" not in winners
+    assert [s.display_name for s in current_streaks(winners)] == ["kuhy"]