#!/usr/bin/env python3 """Re-test published point-level tennis findings on a point-by-point score record. Live Tennis API, "Tennis findings, re-tested": https://blog.livetennisapi.com/tennis-findings-retested Four claims from the research literature, each measured the same way on every match: iid Klaassen and Magnus (2001): points are not independent (winning the previous point helps the server win the next) and not identically distributed (the server does worse on break points). Both effects are small. serve-first Magnus and Klaassen (1999): serving first helps a little in the first set and not after it; in later sets the first server is usually the weaker player. break-back Magnus and Klaassen (1996), hypothesis 12: after breaking, a player is more likely to be broken straight back. They found the opposite. hot-hand Momentum studies of Wimbledon 2023 (Wang, Guo and Zhou 2024; Du, Zhang and Zhou 2025): recent success predicts the next point or game. The point-level effects are measured inside each match, against what the same players would produce if every point were independent at their own serve win rate in that match. That is the comparison the claims are about, and it keeps the strength gap between two players from masquerading as momentum. Requirements: Python 3.10 or newer, standard library only. No numpy, no pandas. Run it on the public sample (Zenodo DOI 10.5281/zenodo.22048730): python retest_findings.py run --point-reps 100 --game-reps 200 \\ --points points_sample_2026-06.csv.gz --matches matches.csv.gz > sample.json The points file holds one row per score state: match_id, games_p1, games_p2 (per-set JSON arrays), points_p1, points_p2, server, is_tiebreak, in capture order. The matches file supplies tier_key, surface, best_of and event_status. A single CSV that carries those match columns on every row can be read with --tape instead (add winner_side when you have it). Licence: MIT for this file. The figures it produced for the page are CC BY 4.0. """ from __future__ import annotations import argparse import csv import gzip import io import json import math import random import re import sys from collections import Counter, defaultdict from dataclasses import dataclass, field VERSION = "1.0.0" # Tier keys of the six singles tours, as they appear in the dataset's tier_key column. TOURS = { "265": "ATP", "266": "WTA", "281": "Challenger M", "272": "Challenger W", "270": "ITF M", "271": "ITF W", } SURFACES = ("hard", "clay", "grass") # Matches that did not finish normally. Compared case-insensitively against event_status. NOT_FINISHED = ("retir", "walk", "w/o", "cancel", "interrupt", "abandon", "default", "postpone") REGULAR_POINTS = {"0": 0, "15": 1, "30": 2, "40": 3, "A": 4, "AD": 4} csv.field_size_limit(1 << 24) # --------------------------------------------------------------------------- scoring rules -- def set_won(a: int, b: int) -> bool: """A finished standard set: 6-0 to 6-4, 7-5, or 7-6 after a tiebreak.""" hi, lo = max(a, b), min(a, b) return (hi == 6 and lo <= 4) or (hi, lo) in ((7, 5), (7, 6)) def set_in_progress(a: int, b: int) -> bool: return 0 <= a <= 6 and 0 <= b <= 6 and not set_won(a, b) def regular_state_legal(a: int, b: int) -> bool: return (a <= 3 and b <= 3) or (a, b) in ((4, 3), (3, 4)) def point_step(prev: tuple, cur: tuple, tiebreak: bool) -> int: """Who won the single point between two states of one game: 1, 2, or 0 if it is not one point.""" (a0, b0), (a1, b1) = prev, cur if (a1, b1) == (a0 + 1, b0): return 1 if (a1, b1) == (a0, b0 + 1): return 2 if not tiebreak: if (a0, b0) == (4, 3) and (a1, b1) == (3, 3): return 2 if (a0, b0) == (3, 4) and (a1, b1) == (3, 3): return 1 return 0 def ends_game(state: tuple, winner: int, tiebreak: bool) -> bool: """True when one more point won by `winner` from `state` ends the game.""" a, b = state if winner == 1 else (state[1], state[0]) if tiebreak: return a + 1 >= 7 and a + 1 - b >= 2 return (a == 3 and b <= 2) or a == 4 def point_total(state: tuple) -> int: return state[0] + state[1] # --------------------------------------------------------------------------- reading a tape -- @dataclass class Pos: done: tuple # finished sets, ((a, b), ...) cur: tuple # games in the set being played tiebreak: bool pts: tuple # in-game score as counts terminal: bool # the match is over at this state gi: int # games played in the match before this state's game def parse_games(text: str): s = (text or "").strip() if not s or s in ("[]", "null", "None"): return [] s = s.strip("[]") return [int(x) for x in s.split(",") if x.strip()] def parse_flag(text) -> bool: return str(text).strip().lower() in ("t", "true", "1", "yes") def parse_server(text): s = str(text or "").strip() return int(s) if s in ("1", "2") else None def make_pos(g1: list, g2: list, p1: str, p2: str, sets_to_win: int = 2): """One score state, or None when the state is not a legal tennis score.""" if len(g1) != len(g2): return None n = len(g1) done: list = [] cur = (0, 0) for i in range(n): a, b = g1[i], g2[i] if set_won(a, b): done.append((a, b)) elif i == n - 1 and set_in_progress(a, b): cur = (a, b) else: return None w1 = sum(1 for a, b in done if a > b) w2 = len(done) - w1 if max(w1, w2) > sets_to_win: return None # A set after the match was already decided is not a score. for k in range(len(done)): before = done[:k] b1 = sum(1 for a, b in before if a > b) if max(b1, len(before) - b1) >= sets_to_win: return None terminal = max(w1, w2) == sets_to_win if terminal and cur != (0, 0): return None tiebreak = cur == (6, 6) if terminal: pts = (0, 0) elif tiebreak: try: pts = (int(p1), int(p2)) except (TypeError, ValueError): return None if pts[0] < 0 or pts[1] < 0: return None else: a = REGULAR_POINTS.get(str(p1).strip().upper()) b = REGULAR_POINTS.get(str(p2).strip().upper()) if a is None or b is None or not regular_state_legal(a, b): return None pts = (a, b) gi = sum(a + b for a, b in done) + cur[0] + cur[1] return Pos(tuple(done), cur, tiebreak, pts, terminal, gi) @dataclass class Match: """What one match's tape proves, in the compact form the studies read.""" tour: str surface: str first_server: int | None # who served game 0 of the match sets: list # finished set scores, (a, b), in order; None when unknown set_winners: list # 1 | 2 | None per set games: dict # game index -> (set_no, tiebreak, winner 1|2|None) points: dict # game index -> tuple of point winners (1|2), complete games only winner: int | None stats: Counter = field(default_factory=Counter) def server_of(first_server: int, gi: int) -> int: """Service alternates every game across the whole match, a tiebreak counting as one game.""" return first_server if gi % 2 == 0 else 3 - first_server def reachable(a: Pos, b: Pos) -> bool: """Can the score move forward from state a to state b (b == a counts)?""" if b.gi != a.gi: if b.gi < a.gi or len(b.done) < len(a.done) or b.done[:len(a.done)] != a.done: return False # the set a was playing must have moved forward on both sides nxt = b.done[len(a.done)] if len(b.done) > len(a.done) else b.cur return nxt[0] >= a.cur[0] and nxt[1] >= a.cur[1] if b.done != a.done or b.cur != a.cur: return False if b.terminal or a.terminal: return b.terminal == a.terminal (x0, y0), (x1, y1) = a.pts, b.pts if not a.tiebreak and min(x0, y0) >= 3 and min(x1, y1) >= 3: return True # deuce, advantage and back again return x1 >= x0 and y1 >= y0 def _progress(a: Pos, b: Pos) -> bool: """Does b come strictly later in the match than a (more games, or more points played in the same game)? Deuce to advantage and back is not progress.""" if b.gi != a.gi: return b.gi > a.gi if a.terminal or b.terminal: return b.terminal and not a.terminal if not a.tiebreak and min(a.pts) >= 3 and min(b.pts) >= 3: return False return point_total(b.pts) > point_total(a.pts) def confirmed(positions: list, window: int = 60) -> list: """Per row: True when it lies on the longest run of rows that the score can actually move through, each reachable from the one before. A row that jumped ahead and was walked back (an early award, a mirrored set, a final that was not one), a stale row sent late, or a stale tail after the final score all fall off that run, however many of them there are, because the true score keeps moving and they do not. Links are looked for up to `window` rows back and to the ends of the eight longest runs so far, which keeps the work linear; a clean record, where every row can follow the one before, skips the search entirely.""" n = len(positions) if all(reachable(positions[i], positions[i + 1]) for i in range(n - 1)): return [True] * n def key(p): return (p.done, p.cur, p.pts, p.terminal) # best = how many steps of real progress the run ending here makes. A repeat, or a swing # between deuce and advantage, adds nothing, so stale copies and replayed deuce rows can # never outscore the run through the play they skipped. best = [1] * n prev = [-1] * n top: list = [] # the rows ending the longest runs so far, longest first for j in range(n): pj = positions[j] kj = key(pj) # The recent rows, plus the ends of the longest runs however far back they are, so a # burst of stale rows longer than the window cannot cut the true run in two. cands = sorted(set(range(max(0, j - window), j)) | set(top)) for i in cands: gain = best[i] + (1 if _progress(positions[i], pj) else 0) if gain >= best[j] and reachable(positions[i], pj): best[j] = gain prev[j] = i # one slot per distinct state, so copies of one bad row cannot fill every slot top = [k for k in top if key(positions[k]) != kj] + [j] top.sort(key=lambda k: (best[k], k), reverse=True) del top[8:] keep = [False] * n j = max(range(n), key=lambda k: (best[k], k)) if n else -1 while j >= 0: keep[j] = True j = prev[j] return keep def build_match(rows, tour: str, surface: str, winner_side=None, sets_to_win: int = 2) -> Match: """Walk one tape forward and keep only what it proves. `rows` is an iterable of (games_p1, games_p2, points_p1, points_p2, server, is_tiebreak) with the games as lists. Rows that move backwards, repeat a state, contradict a set already seen finished, or are not a legal score are skipped, never repaired. """ st = Counter() fixed: dict = {} # set index -> final score gpos: dict = {} # game index -> (set_no, games before the game) states: dict = defaultdict(list) votes: dict = defaultdict(Counter) tainted: set = set() # games where the record went sideways or backwards last = None terminal = None parsed = [] for g1, g2, p1, p2, server, _tb in rows: st["rows"] += 1 pos = make_pos(g1, g2, p1, p2, sets_to_win) if pos is None: st["rows_illegal"] += 1 continue parsed.append((pos, server)) # Rows off the longest run the score can move through were never the score (confirmed()). keep = confirmed([p for p, _ in parsed]) st["rows_unconfirmed"] += len(parsed) - sum(keep) # A dropped row showing a state its game never showed on the kept run means that game's # record is in doubt (the run may have taken a wrong branch through it): no points from it. kept_states: dict = defaultdict(set) for (pos, _s), ok in zip(parsed, keep, strict=True): if ok and not pos.terminal: kept_states[pos.gi].add((pos.done, pos.cur, pos.pts)) doubtful = {pos.gi for (pos, _s), ok in zip(parsed, keep, strict=True) if not ok and not pos.terminal and pos.gi in kept_states and (pos.done, pos.cur, pos.pts) not in kept_states[pos.gi]} tainted |= doubtful # Dropped rows that put a different games score on a game index the kept run uses: the # record branched there. Which branch was the real one is a guess, so the game into that # index and the game at it get no winner (and their sets drop out of the game tests). kept_keys: dict = {} for (pos, _s), ok in zip(parsed, keep, strict=True): if ok: kept_keys.setdefault(pos.gi, (pos.done, pos.cur)) # A stray row in the middle (a mirrored or torn snapshot) is noise: rows on the kept run # speak about the same game before and after it. When the record's first or last word on a # game index is a dropped row with a different games score, it is a competing account. first_word: dict = {} last_word: dict = {} for (pos, _s), ok in zip(parsed, keep, strict=True): if pos.gi in kept_keys: fine = ok or (pos.done, pos.cur) == kept_keys[pos.gi] first_word.setdefault(pos.gi, fine) last_word[pos.gi] = fine disputed = {gi for gi in last_word if not (first_word[gi] and last_word[gi])} for (pos, server), ok in zip(parsed, keep, strict=True): if not ok: continue if any(fixed.get(i, d) != d for i, d in enumerate(pos.done)): st["rows_conflict"] += 1 continue key = (len(pos.done), pos.cur) if gpos.get(pos.gi, key) != key: st["rows_conflict"] += 1 continue if last is not None: if pos.gi < last.gi: # The record went back to an earlier game. A plain re-send of a state we hold # is harmless, but a state that game never showed means the game was left # early (an early award, later corrected): its point record is incomplete. if not pos.terminal and pos.pts not in states.get(pos.gi, ()): tainted.add(pos.gi) continue if pos.gi > last.gi and not reachable(last, pos): # A jump to a later game that no play could reach from here (a score that # goes down, a set that changes): never let it name the games in between. st["rows_conflict"] += 1 continue if pos.gi == last.gi: if pos.terminal or last.terminal or pos.pts == last.pts: if server in (1, 2) and not pos.tiebreak and not pos.terminal: votes[pos.gi][server] += 1 continue if (not point_step(last.pts, pos.pts, pos.tiebreak) and point_total(pos.pts) <= point_total(last.pts)): # A row from the same game that is neither the next point nor ahead: # the game's point record can no longer be trusted. tainted.add(pos.gi) continue for i, d in enumerate(pos.done): fixed.setdefault(i, d) gpos.setdefault(pos.gi, key) if pos.terminal: terminal = pos break states[pos.gi].append(pos.pts) if server in (1, 2) and not pos.tiebreak: votes[pos.gi][server] += 1 last = pos # Game winners, read off the step to the next game we saw. games: dict = {} for gi, (s, cur) in gpos.items(): if terminal is not None and gi == terminal.gi: continue winner = None nxt = gpos.get(gi + 1) if nxt is not None: s2, cur2 = nxt if s2 == s: if cur2 == (cur[0] + 1, cur[1]): winner = 1 elif cur2 == (cur[0], cur[1] + 1): winner = 2 elif s2 == s + 1 and cur2 == (0, 0) and s in fixed: if fixed[s] == (cur[0] + 1, cur[1]): winner = 1 elif fixed[s] == (cur[0], cur[1] + 1): winner = 2 if gi in disputed or gi + 1 in disputed: winner = None games[gi] = (s, cur == (6, 6), winner, cur) # Dropped rows that disagree with a set score the kept run fixed: a different final score # for that set, or a set still being played past the score it supposedly ended on. Which # one is real is a guess, so that set's score and winner stay unknown, and so does a match # winner read off the kept run. # The same first-or-last-word rule per set. set_first: dict = {} set_last: dict = {} def _word(i, fine): set_first.setdefault(i, fine) set_last[i] = fine for (pos, _s), ok in zip(parsed, keep, strict=True): for i, d in enumerate(pos.done): if i in fixed: _word(i, ok or fixed[i] == d) k = len(pos.done) if k in fixed and not pos.terminal: _word(k, ok or (pos.cur[0] <= fixed[k][0] and pos.cur[1] <= fixed[k][1])) disputed_sets = {i for i in set_last if not (set_first[i] and set_last[i])} match_winner = None if terminal is not None and not disputed_sets: w1 = sum(1 for a, b in terminal.done if a > b) match_winner = 1 if w1 == sets_to_win else 2 # A tape that stops before the final score proves nothing about the last game: match point # can be saved and more games played. That game and that set stay unknown; the match winner # comes from the recorded result when there is one. if match_winner is None and winner_side in (1, 2): match_winner = winner_side if winner_side in (1, 2) and match_winner not in (None, winner_side): st["winner_disagrees"] += 1 return Match(tour, surface, None, [], [], {}, {}, None, st) # Point sequences of complete games: 0-0, single points all the way, a legal last point. # Tiebreaks are left out: no test reads their points (the server changes inside them), and # a 10-point deciding tiebreak would need its own end rule. points: dict = {} for gi, (_s, tb, winner, _cur) in games.items(): seq = states.get(gi) if tb or not seq or winner is None or seq[0] != (0, 0) or gi in tainted: continue out = [] ok = True for prev, cur in zip(seq, seq[1:], strict=False): w = point_step(prev, cur, tb) if not w: ok = False break out.append(w) if ok and ends_game(seq[-1], winner, tb): out.append(winner) points[gi] = tuple(out) # Who served first. Each non-tiebreak game with a clear majority server votes, through # the alternation rule; a match whose votes disagree is not used for anything that needs # the server. s0_votes = Counter() for gi, c in votes.items(): if gi in games and games[gi][1]: continue top = c.most_common(2) if not top or (len(top) == 2 and top[0][1] == top[1][1]): continue m = top[0][0] s0_votes[m if gi % 2 == 0 else 3 - m] += 1 first_server = None total = sum(s0_votes.values()) if total >= 3: cand, n = s0_votes.most_common(1)[0] if n / total >= 0.9: first_server = cand if first_server is None: st["server_unresolved"] += 1 nsets = (max(fixed) + 1) if fixed else 0 sets = [None if i in disputed_sets else fixed.get(i) for i in range(nsets)] set_winners = [None if sc is None else (1 if sc[0] > sc[1] else 2) for sc in sets] if match_winner in (1, 2): decided = sum(1 for w in set_winners if w is not None) w_sets = sum(1 for w in set_winners if w == match_winner) if (decided == len(set_winners) and w_sets == sets_to_win - 1 and decided - w_sets == sets_to_win - 1): # The deciding set itself is missing from the record; its winner is not. set_winners.append(match_winner) sets.append(None) out_games = {gi: (s, tb, w) for gi, (s, tb, w, _c) in games.items()} st["games"] += len(out_games) st["games_with_winner"] += sum(1 for v in out_games.values() if v[2]) st["games_with_points"] += len(points) return Match(tour, surface, first_server, sets, set_winners, out_games, points, match_winner, st) def match_quality(m: Match) -> tuple: return (m.first_server is not None, sum(1 for v in m.games.values() if v[2]), len(m.points)) # --------------------------------------------------------------------------- readers -- def _open(path: str): if path == "-": return io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8", newline="") if path.endswith(".gz"): return gzip.open(path, "rt", encoding="utf-8", newline="") return open(path, encoding="utf-8", newline="") def _row(r: dict): return (parse_games(r.get("games_p1")), parse_games(r.get("games_p2")), r.get("points_p1"), r.get("points_p2"), parse_server(r.get("server")), parse_flag(r.get("is_tiebreak"))) def eligible(meta: dict): """(tour, surface) for a completed best-of-three singles match, else None.""" tour = TOURS.get(str(meta.get("tier_key") or "").split("_")[0]) if tour is None: return None fmt = str(meta.get("best_of") or "").strip().upper() if fmt != "BO3": return None status = str(meta.get("event_status") or "").lower() if any(word in status for word in NOT_FINISHED): return None surface = str(meta.get("surface") or "").strip().lower() return tour, (surface if surface in SURFACES else "") def read_split(points_path: str, matches_path: str, counts: Counter): """The Zenodo layout: a points file plus a matches file, joined on match_id.""" meta = {} with _open(matches_path) as f: for r in csv.DictReader(f): meta[r["match_id"]] = r tapes = defaultdict(list) with _open(points_path) as f: for i, r in enumerate(csv.DictReader(f)): ts = r.get("timestamp_utc") or "" tapes[r["match_id"]].append((ts, i, r)) for mid, items in tapes.items(): counts["tapes"] += 1 info = meta.get(mid) if info is None: counts["no_match_row"] += 1 continue el = eligible(info) if el is None: counts["not_eligible"] += 1 continue items.sort(key=lambda t: (t[0], t[1])) ws = parse_server(info.get("winner_side")) yield mid, el, ws, [[_row(r) for _, _, r in items]] def read_tape(path: str, counts: Counter): """One CSV with the match columns (tier_key, surface, best_of, event_status and, when known, winner_side) on every row, grouped by match_id, rows in score order.""" from itertools import groupby with _open(path) as f: reader = csv.DictReader(f) for mid, grp in groupby(reader, key=lambda r: r["match_id"]): counts["tapes"] += 1 grp = list(grp) el = eligible(grp[0]) if el is None: counts["not_eligible"] += 1 continue ws = parse_server(grp[0].get("winner_side")) yield mid, el, ws, [[_row(r) for r in grp]] def load_matches(source, counts: Counter) -> list: out = [] for _mid, (tour, surface), ws, sequences in source: best = None for rows in sequences: m = build_match(rows, tour, surface, ws) if best is None or match_quality(m) > match_quality(best): best = m counts.update({f"walk_{k}": v for k, v in best.stats.items()}) if not best.games: counts["no_usable_games"] += 1 continue best.stats = Counter() out.append(best) counts["matches"] = len(out) return out # --------------------------------------------------------------------------- estimators -- class Ratio: """A ratio of sums over matches, with a confidence interval clustered by match. Each match adds (num, den) for the observed data and, where a null applies, the null's expected (num, den) for that same match. With a null, the reported effect is observed minus null, and its interval comes from the linearised (delta-method) variance of that difference, with each match's observed and null parts kept together, so their covariance counts. The running sums of products of the four parts are enough for that. """ __slots__ = ("m", "units", "s", "ss") def __init__(self): self.m = 0 self.units = 0 self.s = [0.0] * 4 # observed num, den; null num, den self.ss = [[0.0] * 4 for _ in range(4)] def add(self, num: float, den: float, units: int = 0, null=None): if den == 0 and units == 0 and (null is None or null[1] == 0): return self.m += 1 self.units += units v = (num, den) + (tuple(null) if null is not None else (0.0, 0.0)) for i in range(4): self.s[i] += v[i] for j in range(4): self.ss[i][j] += v[i] * v[j] @property def sn(self): return self.s[0] @property def sd(self): return self.s[1] def _var(self, coef) -> float: return sum(coef[i] * coef[j] * self.ss[i][j] for i in range(4) for j in range(4)) def result(self, scale: float = 100.0, null: bool = False) -> dict: sn, sd, zn, zd = self.s if sd <= 0: return {"units": self.units, "matches": self.m, "value_pct": None} r = sn / sd out = {"units": self.units, "matches": self.m} if null and zd > 0: r0 = zn / zd ex = r - r0 se = math.sqrt(max(self._var((1 / sd, -r / sd, -1 / zd, r0 / zd)), 0.0)) out.update({"observed_pp": round(r * scale, 2), "null_pp": round(r0 * scale, 2), "excess_pp": round(ex * scale, 2), "ci95_pp": [round((ex - 1.96 * se) * scale, 2), round((ex + 1.96 * se) * scale, 2)]}) else: se = math.sqrt(max(self._var((1 / sd, -r / sd, 0.0, 0.0)), 0.0)) out.update({"value_pct": round(r * scale, 2), "ci95_pct": [round((r - 1.96 * se) * scale, 2), round((r + 1.96 * se) * scale, 2)]}) half = 1.96 * se * scale out["thin"] = bool(half > 2.0 or self.units < 500) return out class Diff: """Difference of two ratios measured on the same matches (A minus B), clustered by match.""" def __init__(self): self.a, self.b = Ratio(), Ratio() self.rows = [] def add(self, na, da, nb, db_): if da == 0 and db_ == 0: return self.a.add(na, da, int(da)) self.b.add(nb, db_, int(db_)) self.rows.append((na, da, nb, db_)) def result(self, scale: float = 100.0) -> dict: if self.a.sd <= 0 or self.b.sd <= 0: return {"units": self.a.units + self.b.units, "matches": len(self.rows), "difference_pp": None} ra, rb = self.a.sn / self.a.sd, self.b.sn / self.b.sd var = 0.0 for na, da, nb, db_ in self.rows: z = (na - ra * da) / self.a.sd - (nb - rb * db_) / self.b.sd var += z * z se = math.sqrt(var) d = ra - rb half = 1.96 * se * scale return {"units": self.a.units + self.b.units, "matches": len(self.rows), "a_pct": round(ra * scale, 2), "a_units": self.a.units, "b_pct": round(rb * scale, 2), "b_units": self.b.units, "difference_pp": round(d * scale, 2), "ci95_pp": [round((d - 1.96 * se) * scale, 2), round((d + 1.96 * se) * scale, 2)], "thin": bool(half > 2.0 or min(self.a.units, self.b.units) < 500)} def groups_of(m: Match) -> list: g = ["all", "tour:" + m.tour] if m.surface: g.append("surface:" + m.surface) return g def mh(n_a: int, w_a: int, n_b: int, w_b: int) -> tuple: """Mantel-Haenszel risk-difference contribution of one stratum: P(win|A) - P(win|B).""" t = n_a + n_b if n_a == 0 or n_b == 0: return 0.0, 0.0 return (w_a * n_b - w_b * n_a) / t, n_a * n_b / t # --------------------------------------------------------------------------- study 1 & 4a -- # Service points inside complete, non-tiebreak service games. One stratum per (match, server). # The null reshuffles the points inside each game (reshuffle_game). POINT_KEYS = ("prev1", "prev2", "prev3", "breakpoint") def service_point_counts(games: list) -> dict: """Counts behind the point-level tests for one server's games (lists of True = server won). Only points played before the game first reaches deuce are counted. In a live record the score can flicker between deuce and advantage without a point being played, and nothing in the record tells a real deuce battle from a flicker; leaving the deuce zone out removes the problem for every source alike (the simulated null follows the same rule). prevK: points with at least K earlier points in the same game, split by whether the server won all K of them. breakpoint: every counted point, split by whether the receiver was one point from the game (0-40, 15-40, 30-40). Break points follow a lost point more often than other points do, so that split is made separately after a won point, after a lost point and on a game's first point, then combined. That removes the obvious overlap but not all of it: the null shuffles points inside a game, which breaks any dependence between points, so a previous-point effect alone also moves the break-point figure. Read the two together; the break-point figure on its own is a test of iid, not of pressure. Each entry is a list of strata, each stratum [n_a, w_a, n_b, w_b]. """ c = {"prev1": [[0, 0, 0, 0]], "prev2": [[0, 0, 0, 0]], "prev3": [[0, 0, 0, 0]], "breakpoint": [[0, 0, 0, 0], [0, 0, 0, 0], [0, 0, 0, 0]]} won = total = bp_faced = bp_saved = 0 for g in games: a = b = 0 for j, w in enumerate(g): if a >= 3 and b >= 3: break # deuce reached: the rest of the game is not counted total += 1 won += w for k in (1, 2, 3): if j >= k: row = c[f"prev{k}"][0] if all(g[j - k:j]): row[0] += 1 row[1] += w else: row[2] += 1 row[3] += w row = c["breakpoint"][0 if j == 0 else (1 if g[j - 1] else 2)] if b == 3: row[0] += 1 row[1] += w bp_faced += 1 bp_saved += w else: row[2] += 1 row[3] += w if w: a += 1 else: b += 1 c["_points"] = (total, won, bp_faced, bp_saved) return c def reshuffle_game(game: tuple, rnd) -> list: """The same game with its counted points in a random legal order. Under independence with a constant chance, every order of a game's points that is a legal game is equally likely once you know how many points each side won. So shuffling inside the game is the exact null, and it keeps the game's length and score. That matters because a game only reaches these tests when every one of its points is in the record, and longer games are dropped more often: a null that generated fresh games would not share that selection, and the selection alone produces "momentum". A game that reached deuce has its first six points shuffled (three each, any order is legal) and the rest left, since the tests stop counting at deuce. Any other game has every point but the last shuffled; the last one is always the winner's. """ g = list(game) a = b = 0 deuce_at = None for i, w in enumerate(g): if w: a += 1 else: b += 1 if a >= 3 and b >= 3: deuce_at = i + 1 break n = deuce_at if deuce_at is not None else len(g) - 1 head = g[:n] for i in range(len(head) - 1, 0, -1): j = int(rnd() * (i + 1)) head[i], head[j] = head[j], head[i] return head + g[n:] def service_sequences(m: Match) -> dict: """server -> list of complete service games as tuples of bool (server won the point).""" seqs = defaultdict(list) if m.first_server is None: return seqs for gi, pts in m.points.items(): if m.games[gi][1]: continue srv = server_of(m.first_server, gi) seqs[srv].append(tuple(w == srv for w in pts)) return seqs def study_points(matches: list, reps: int, rnd) -> dict: stats = {k: defaultdict(Ratio) for k in POINT_KEYS} raw = defaultdict(lambda: [0, 0, 0, 0, 0]) # points, won, bp faced, bp saved, games pooled = {k: defaultdict(lambda: [0, 0, 0, 0]) for k in POINT_KEYS} for m in matches: seqs = service_sequences(m) if not seqs: continue obs = {k: [0.0, 0.0] for k in POINT_KEYS} nul = {k: [0.0, 0.0] for k in POINT_KEYS} units = {k: 0 for k in POINT_KEYS} tot = [0, 0, 0, 0, 0] pool = {k: [0, 0, 0, 0] for k in POINT_KEYS} for games in seqs.values(): c = service_point_counts(games) total, won, bpf, bps = c["_points"] tot[0] += total tot[1] += won tot[2] += bpf tot[3] += bps tot[4] += len(games) for k in POINT_KEYS: for stratum in c[k]: num, den = mh(*stratum) obs[k][0] += num obs[k][1] += den units[k] += stratum[0] + stratum[2] for i in range(4): pool[k][i] += stratum[i] if total == 0: continue for _ in range(reps): cs = service_point_counts([reshuffle_game(g, rnd) for g in games]) for k in POINT_KEYS: for stratum in cs[k]: num, den = mh(*stratum) nul[k][0] += num / reps nul[k][1] += den / reps for g in groups_of(m): for i in range(5): raw[g][i] += tot[i] for k in POINT_KEYS: stats[k][g].add(obs[k][0], obs[k][1], units[k], nul[k]) for i in range(4): pooled[k][g][i] += pool[k][i] out = {} for k in POINT_KEYS: out[k] = {} for g, r in stats[k].items(): res = r.result(null=True) na, wa, nb, wb = pooled[k][g] res["pooled_a_pct"] = round(100 * wa / na, 2) if na else None res["pooled_b_pct"] = round(100 * wb / nb, 2) if nb else None res["pooled_a_units"], res["pooled_b_units"] = na, nb out[k][g] = res out["serve_points"] = { g: {"points_before_deuce": v[0], "service_games": v[4], "serve_points_won_pct": round(100 * v[1] / v[0], 2) if v[0] else None, "break_points_before_deuce": v[2], "break_points_saved_pct": round(100 * v[3] / v[2], 2) if v[2] else None} for g, v in raw.items()} return out # --------------------------------------------------------------------------- study 2 -- def study_serve_first(matches: list) -> dict: first = defaultdict(lambda: defaultdict(Ratio)) # set number -> group -> ratio match_first = defaultdict(Ratio) cond = defaultdict(Diff) # previous-set winner, sets 2 and 3 cond_by_set = defaultdict(lambda: defaultdict(Diff)) prev_winner_serves = defaultdict(Ratio) for m in matches: if m.first_server is None or not m.set_winners: continue per_set = {} gi = 0 for k, w in enumerate(m.set_winners): if w is None: break fs = server_of(m.first_server, gi) per_set[k] = (fs, w) sc = m.sets[k] if k < len(m.sets) else None if sc is None: break gi += sc[0] + sc[1] # a 7-6 set is 13 games: the tiebreak counts as one groups = groups_of(m) for k, (fs, w) in per_set.items(): for g in groups: first[k + 1][g].add(1.0 if fs == w else 0.0, 1.0, 1) if 0 in per_set and m.winner in (1, 2): for g in groups: match_first[g].add(1.0 if per_set[0][0] == m.winner else 0.0, 1.0, 1) tot = [0, 0, 0, 0] serves_first = [0, 0] # later sets the previous set's winner served first, of all by_set = {} for k in (1, 2): if k not in per_set or (k - 1) not in per_set: continue fs, w = per_set[k] pw = per_set[k - 1][1] serves = fs == pw won = 1 if w == pw else 0 row = [0, 0, 0, 0] if serves: row[0], row[1] = 1, won else: row[2], row[3] = 1, won by_set[k + 1] = row for i in range(4): tot[i] += row[i] serves_first[0] += serves serves_first[1] += 1 for g in groups: if serves_first[1]: # sets 2 and 3 of one match are one unit for the interval, not two prev_winner_serves[g].add(serves_first[0], serves_first[1], serves_first[1]) if tot[0] or tot[2]: cond[g].add(tot[1], tot[0], tot[3], tot[2]) for sn, row in by_set.items(): cond_by_set[sn][g].add(row[1], row[0], row[3], row[2]) return { "first_server_wins_set": {str(k): {g: r.result() for g, r in v.items()} for k, v in sorted(first.items())}, "first_server_in_set_1_wins_match": {g: r.result() for g, r in match_first.items()}, "previous_set_winner_serves_first": {g: r.result() for g, r in prev_winner_serves.items()}, "previous_set_winner_wins_serving_vs_receiving": {g: d.result() for g, d in cond.items()}, "previous_set_winner_by_set": {str(k): {g: d.result() for g, d in v.items()} for k, v in sorted(cond_by_set.items())}, } # --------------------------------------------------------------------------- studies 3 & 4b -- # Game sequences. One stratum per (match, player, serving or receiving). The comparison is # the player's chance of winning a game after winning the previous K games of the set # against their chance in the same role otherwise. Inside one match that comparison is biased # even when every game is independent (a run of breaks uses up the breaks the match had), so # the null shuffles each player's own service-game results among their service games in the # same set, keeping only orders under which the set still ends where it ended, and recomputes # it. That is the exact null for independent games given each set's holds and breaks. def game_rows(m: Match) -> list: """Per game index: (set_no, tiebreak, server or 0, winner or 0); None for an unseen game.""" if m.first_server is None or not m.games: return [] n = max(m.games) + 1 out = [] for gi in range(n): v = m.games.get(gi) if v is None: out.append(None) continue s, tb, w = v out.append((s, tb, 0 if tb else server_of(m.first_server, gi), w or 0)) return out GAME_KEYS = ("k1", "serve_k2", "return_k2", "serve_k3", "return_k3") def game_strata(rows: list) -> dict: """key -> {(player, role): [n_a, w_a, n_b, w_b]}, a = the player won the previous K games of this set, b = they did not. k1 is the same comparison from either side of the net, so it is counted once, from the server's side: it is the break-back test.""" c = {k: {} for k in GAME_KEYS} for i, g in enumerate(rows): if g is None or g[1] or not g[3]: continue s, _tb, srv, w = g for k in (1, 2, 3): if i < k: break prev = rows[i - k:i] if any(q is None or q[0] != s or not q[3] for q in prev): break for player in ((srv,) if k == 1 else (1, 2)): role = "serve" if player == srv else "return" key = "k1" if k == 1 else f"{role}_k{k}" row = c[key].setdefault((player, role), [0, 0, 0, 0]) won = w == player if all(q[3] == player for q in prev): row[0] += 1 row[1] += won else: row[2] += 1 row[3] += won return c def _set_order_legal(games: list) -> bool: """True when a set's game results, in this order, never end the set before its last game (and, for a set the record shows unfinished, never end it at all).""" a = b = 0 finished = None for k, (_tb, w) in enumerate(games): if w == 1: a += 1 else: b += 1 if set_won(a, b): finished = k break if finished is None: return True return finished == len(games) - 1 def shuffleable_sets(rows: list) -> set: """Sets whose first game and every game are known: the only ones the null can shuffle.""" by_set: dict = defaultdict(list) for i, g in enumerate(rows): if g is not None: by_set[g[0]].append(i) out = set() for s_no, idx in by_set.items(): first, last = idx[0], idx[-1] start_known = first == 0 or (rows[first - 1] is not None and rows[first - 1][0] != s_no) if start_known and all(g is not None and g[3] for g in rows[first:last + 1]): out.add(s_no) return out def shuffled(rows: list, rnd, tries: int = 100) -> list: """The same match with each player's service-game results permuted among their own service games in the same set, keeping only orders under which every set ends where it really ended. A set whose games are not all known is left as it is.""" out = list(rows) by_set: dict = defaultdict(list) for i, g in enumerate(rows): if g is not None: by_set[g[0]].append(i) usable = shuffleable_sets(rows) for s_no, idx in by_set.items(): if s_no not in usable: continue slots = {srv: [i for i in idx if not rows[i][1] and rows[i][2] == srv] for srv in (1, 2)} for _ in range(tries): trial = {i: rows[i][3] for i in idx} for pos in slots.values(): res = [rows[i][3] for i in pos] for k in range(len(res) - 1, 0, -1): j = int(rnd() * (k + 1)) res[k], res[j] = res[j], res[k] trial.update(zip(pos, res, strict=True)) if _set_order_legal([(rows[i][1], trial[i]) for i in idx]): for i in idx: s, tb, sv, _w = rows[i] out[i] = (s, tb, sv, trial[i]) break return out def _mh_sum(strata: dict) -> tuple: num = den = 0.0 units = 0 for row in strata.values(): a, b = mh(*row) num += a den += b units += row[0] + row[2] return num, den, units def study_games(matches: list, reps: int, rnd) -> dict: est = {k: defaultdict(Ratio) for k in GAME_KEYS} pooled = {k: defaultdict(lambda: [0, 0, 0, 0]) for k in GAME_KEYS} hold_all = defaultdict(Ratio) after_break_vs_all = defaultdict(Diff) for m in matches: rows = game_rows(m) if not rows: continue groups = groups_of(m) # Only sets the null can shuffle enter the comparisons, so that observed and null are # computed on the same games. The hold rates below still use every known game. usable = shuffleable_sets(rows) rows_all = rows rows = [g if g is not None and g[0] in usable else None for g in rows_all] obs = game_strata(rows) null = {k: [0.0, 0.0] for k in GAME_KEYS} for _ in range(reps): sh = game_strata([g if g is not None and g[0] in usable else None for g in shuffled(rows_all, rnd)]) for k in GAME_KEYS: num, den, _u = _mh_sum(sh[k]) null[k][0] += num / reps null[k][1] += den / reps n_all = h_all = 0 for g in rows_all: if g is not None and not g[1] and g[3]: n_all += 1 h_all += g[3] == g[2] # The paper's raw comparison reads every known after-break game, not only the sets # the null can shuffle. k1 = [sum(r[i] for r in game_strata(rows_all)["k1"].values()) for i in range(4)] for g in groups: for k in GAME_KEYS: num, den, units = _mh_sum(obs[k]) est[k][g].add(num, den, units, tuple(null[k])) for row in obs[k].values(): for i in range(4): pooled[k][g][i] += row[i] hold_all[g].add(h_all, n_all, n_all) if n_all: # every match counts toward "all service games", with or without a break after_break_vs_all[g].add(k1[1], k1[0], h_all, n_all) out = {} for k in GAME_KEYS: out[k] = {} for g, r in est[k].items(): res = r.result(null=True) na, wa, nb, wb = pooled[k][g] res["pooled_a_pct"] = round(100 * wa / na, 2) if na else None res["pooled_b_pct"] = round(100 * wb / nb, 2) if nb else None res["pooled_a_units"], res["pooled_b_units"] = na, nb out[k][g] = res out["hold_all_service_games"] = {g: r.result() for g, r in hold_all.items()} out["hold_after_breaking_vs_all_pooled"] = {g: d.result() for g, d in after_break_vs_all.items()} return out # --------------------------------------------------------------------------- the claims -- # What each study re-tests, in the words of the paper where we could read it. CLAIMS = [ { "id": "points-iid", "title": "Are points independent and identically distributed?", "claim": ("Points are neither independent nor identically distributed. Winning the " "previous point has a positive effect on winning the current point. At " "important points it is more difficult for the server to win the point. " "The deviations are small, so the iid model is still a good approximation in " "many cases."), "sources": [{ "authors": "Franc J. G. M. Klaassen and Jan R. Magnus", "year": 2001, "title": ("Are Points in Tennis Independent and Identically Distributed? Evidence " "From a Dynamic Binary Panel Data Model"), "journal": "Journal of the American Statistical Association", "volume": "96", "issue": "454", "pages": "500-509", "doi": "10.1198/016214501753168217", "data": "Wimbledon 1992-1995, almost 90,000 points", }], "measures": ["prev1", "prev2", "prev3", "breakpoint"], }, { "id": "serve-first", "title": "Is there an advantage in serving first in a set?", "claim": ("In the men's singles, the first server won 54.7% of first sets. The authors " "named an advantage in serving first in the first set as the one result " "that held up. The women's data showed no clear edge. In later sets the " "first server won less than half the time. The authors put that down to " "who serves first. The set " "winner usually closes the set on their own serve, so the weaker player " "tends to serve first in the next set."), "sources": [{ "authors": "Jan R. Magnus and Franc J. G. M. Klaassen", "year": 1999, "title": "On the Advantage of Serving First in a Tennis Set: Four Years at Wimbledon", "journal": "Journal of the Royal Statistical Society: Series D (The Statistician)", "volume": "48", "issue": "2", "pages": "247-256", "doi": "10.1111/1467-9884.00186", "data": "Wimbledon 1992-1995, almost 90,000 points", }, { "authors": "Jan R. Magnus and Franc J. G. M. Klaassen", "year": 1996, "title": "Testing some common tennis hypotheses: Four years at Wimbledon", "journal": ("CentER Discussion Paper 1996-73, Tilburg University " "(hypothesis 2, the figures quoted here)"), "url": "https://pure.uva.nl/ws/files/2081783/126707_HYPCDP.pdf", "link_text": "full text (PDF, University of Amsterdam)", "data": "481 Wimbledon singles matches, 1992-1995", }], "measures": ["first_server_wins_set", "previous_set_winner_wins_serving_vs_receiving"], }, { "id": "break-back", "title": "After breaking serve, are you more likely to be broken back?", "claim": ("Hypothesis 12 of seventeen commonly heard statements: after breaking your " "opponent's service there is an increased chance that you will lose your " "own service. The authors found the opposite. After a break the hold rate " "rose from 81.8% to 83.3% in the men's singles. In the women's singles it " "rose from 64.3% to 68.3%."), "sources": [{ "authors": "Jan R. Magnus and Franc J. G. M. Klaassen", "year": 1996, "title": "Testing some common tennis hypotheses: Four years at Wimbledon", "journal": "CentER Discussion Paper 1996-73, Tilburg University", "url": "https://pure.uva.nl/ws/files/2081783/126707_HYPCDP.pdf", "link_text": "full text (PDF, University of Amsterdam)", "data": "481 Wimbledon singles matches, 1992-1995", }], "measures": ["k1", "hold_after_breaking_vs_all_pooled"], }, { "id": "hot-hand", "title": "Do streaks predict the next point or game?", "claim": ("Momentum is real: recent success raises the chance of winning the next " "point. The two 2024-2025 papers model the points of the 2023 Wimbledon " "men's singles and report statistical evidence for momentum. Jackson and " "Mosurski asked the older form of the question about whole sets: are heavy " "defeats a sign of psychological momentum or a random effect?"), "sources": [ {"authors": "Jingya Wang, Sihang Guo and Yuanyun Zhou", "year": 2024, "title": ("A multidimensional momentum chain model for tennis matches based on " "difference equations"), "journal": "PLOS ONE", "volume": "19", "issue": "12", "pages": "e0316542", "doi": "10.1371/journal.pone.0316542", "data": "Wimbledon 2023 men's singles, 7,284 points"}, {"authors": "Chang Du, Caiya Zhang and Likai Zhou", "year": 2025, "title": ("A novel methodological framework for analyzing the momentum effect in " "tennis singles"), "journal": "arXiv preprint 2509.01243", "url": "https://arxiv.org/abs/2509.01243", "link_text": "arXiv:2509.01243", "data": "Wimbledon 2023 men's singles"}, {"authors": "David Jackson and Krzysztof Mosurski", "year": 1997, "title": "Heavy Defeats in Tennis: Psychological Momentum or Random Effect?", "journal": "CHANCE", "volume": "10", "issue": "2", "pages": "27-34", "doi": "10.1080/09332480.1997.10542019", "data": "match results by set (see the paper)"}, ], "measures": ["prev2", "prev3", "serve_k2", "return_k2", "serve_k3", "return_k3"], }, ] _FIELD = re.compile(r"\{(full|sample)/([^}|]+)(?:\|(\w+))?\}") def fill(text: str, full: dict, sample: dict) -> str: """Put figures into prose: {full/points/prev1/all/excess_pp|signed} and the like. The page text is written once, with the figures as references, so a new run changes the numbers in the sentences together with the numbers in the tables.""" runs = {"full": full, "sample": sample} def one(m): node = runs[m.group(1)] for part in m.group(2).split("/"): node = node[int(part)] if isinstance(node, list) else node[part] how = m.group(3) or "plain" if how in ("signed", "signed2"): nd = 2 if how == "signed2" else 1 v = round(node, nd) return f"{0:.{nd}f}" if v == 0 else f"{v:+.{nd}f}" if how in ("abs", "abs2"): return f"{abs(node):.{2 if how == 'abs2' else 1}f}" if how == "pct": return f"{node:.1f}%" if how == "int": return f"{int(node):,}" return str(node) return _FIELD.sub(one, text) def _fill_all(obj, full: dict, sample: dict): if isinstance(obj, str): return fill(obj, full, sample) if isinstance(obj, list): return [_fill_all(x, full, sample) for x in obj] if isinstance(obj, dict): return {k: _fill_all(v, full, sample) for k, v in obj.items()} return obj def check_claims(notes: dict, full: dict, sample: dict) -> list: """Every direction the prose asserts, tested against the figures it describes. Each study in the notes lists checks such as {"path": "full/points/prev1/all", "expect": "zero"}. zero: the 95% interval includes 0. positive or negative: the interval lies wholly on that side. includes50 or below50: for a plain percentage. A new run that turns any of them over stops the build, so the words get rewritten with the numbers.""" runs = {"full": full, "sample": sample} bad = [] for sid, st in notes["studies"].items(): for chk in st.get("checks", []): run, *path = chk["path"].split("/") node = runs[run] for part in path: node = node[part] expect = chk["expect"] if "ci95_pp" in node: lo, hi = node["ci95_pp"] ok = {"zero": lo <= 0 <= hi, "positive": lo > 0, "negative": hi < 0}.get(expect) else: lo, hi = node["ci95_pct"] ok = {"includes50": lo <= 50 <= hi, "below50": hi < 50}.get(expect) if not ok: bad.append(f"{sid}: {chk['path']} is [{lo}, {hi}], the text says {expect}") return bad def public_run(run: dict) -> dict: """A run as published: the reading counters under plain names.""" out = dict(run) reading = dict(run.get("reading", {})) if "tapes" in reading: reading["matches_read"] = reading.pop("tapes") out["reading"] = {k.replace("walk_", "record_"): v for k, v in sorted(reading.items())} return out def assemble(full: dict, sample: dict, notes: dict) -> dict: """The published file: the claims, our verdicts, and both runs' figures.""" bad = check_claims(notes, full, sample) if bad: raise SystemExit("the page text no longer matches the figures:\n " + "\n ".join(bad)) notes = _fill_all(notes, full, sample) studies = [] for c in CLAIMS: n = notes["studies"][c["id"]] studies.append({**c, "method": n["method"], "verdict": n["verdict"], "finding": n["finding"], "differs": n.get("differs", ""), "headline": n["headline"]}) return {**notes["meta"], "script_version": VERSION, "studies": studies, "results": {"full": public_run(full), "public_sample": public_run(sample)}} # --------------------------------------------------------------------------- driver -- def corpus_summary(matches: list, counts: Counter) -> dict: by = Counter() surf = Counter() pts = games = 0 with_server = 0 hold = Counter() # all known service games vs the ones with a complete point record for m in matches: by[m.tour] += 1 surf[m.surface or "unrecorded"] += 1 games += sum(1 for v in m.games.values() if v[2]) pts += sum(len(p) for p in m.points.values()) with_server += m.first_server is not None if m.first_server is None: continue for gi, (_s, tb, w) in m.games.items(): if tb or not w: continue held = w == server_of(m.first_server, gi) hold["all"] += 1 hold["all_held"] += held if gi in m.points: hold["complete"] += 1 hold["complete_held"] += held return {"matches_read": counts.get("tapes", 0), "matches_not_eligible": counts.get("not_eligible", 0), "matches_used": len(matches), "matches_with_server_order": with_server, "games_with_known_winner": games, "points_in_complete_games": pts, # If the games with a complete point record were a biased pick, their hold rate would # drift from the hold rate of all games. Published so a reader can check. "hold_pct_all_service_games": round(100 * hold["all_held"] / hold["all"], 2) if hold["all"] else None, "hold_pct_games_with_complete_points": round(100 * hold["complete_held"] / hold["complete"], 2) if hold["complete"] else None, "service_games_with_complete_points": hold["complete"], "service_games_known": hold["all"], "by_tour": dict(sorted(by.items())), "by_surface": dict(sorted(surf.items()))} def analyse(matches: list, counts: Counter, seed: int, point_reps: int, game_reps: int) -> dict: rnd = random.Random(seed).random # noqa: S311 (simulation, not security) return { "script_version": VERSION, "seed": seed, "point_null_reps": point_reps, "game_null_reps": game_reps, "corpus": corpus_summary(matches, counts), "points": study_points(matches, point_reps, rnd), "serve_first": study_serve_first(matches), "games": study_games(matches, game_reps, rnd), } def main(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) sub = ap.add_subparsers(dest="cmd", required=True) run = sub.add_parser("run", help="read a tape and print the results as JSON") src = run.add_mutually_exclusive_group(required=True) src.add_argument("--points", help="points CSV (Zenodo layout); needs --matches") src.add_argument("--tape", help="one CSV with the match columns on every row; - for stdin") run.add_argument("--matches", help="matches CSV (Zenodo layout)") run.add_argument("--seed", type=int, default=20261028) run.add_argument("--point-reps", type=int, default=20, help="simulated matches per real one for the point-level null") run.add_argument("--game-reps", type=int, default=50, help="shuffles per match for the game-level null") run.add_argument("--label", default="") run.add_argument("--save-matches", help="also write the parsed matches here (pickle)") src.add_argument("--load-matches", help="analyse matches saved by --save-matches") asm = sub.add_parser("assemble", help="build the published retest.json from two runs") asm.add_argument("--full", required=True) asm.add_argument("--sample", required=True) asm.add_argument("--notes", required=True) args = ap.parse_args(argv) if args.cmd == "assemble": def load(path): with open(path, encoding="utf-8") as f: return json.load(f) out = assemble(load(args.full), load(args.sample), load(args.notes)) json.dump(out, sys.stdout, indent=1, ensure_ascii=False) sys.stdout.write("\n") return 0 counts = Counter() if args.load_matches: import pickle with open(args.load_matches, "rb") as f: matches, counts = pickle.load(f) # noqa: S301 (a file this script wrote itself) else: if args.points: if not args.matches: ap.error("--points needs --matches") source = read_split(args.points, args.matches, counts) else: source = read_tape(args.tape, counts) matches = load_matches(source, counts) if args.save_matches: import pickle with open(args.save_matches, "wb") as f: pickle.dump((matches, counts), f, protocol=pickle.HIGHEST_PROTOCOL) print(f"read {counts['tapes']} tapes, using {len(matches)} matches", file=sys.stderr) out = analyse(matches, counts, args.seed, args.point_reps, args.game_reps) out["label"] = args.label out["reading"] = {k: v for k, v in sorted(counts.items())} json.dump(out, sys.stdout, indent=1, sort_keys=True) sys.stdout.write("\n") return 0 if __name__ == "__main__": sys.exit(main())