Do substitutions work? Changes made while losing, like for like

mkdir -p ~/statsbomb-open-data && curl -sL https://codeload.github.com/statsbomb/open-data/tar.gz/refs/heads/master \
  | tar xz -C ~/statsbomb-open-data --strip-components=1 --wildcards 'open-data-master/data/competitions.json' \
    'open-data-master/data/matches/*' 'open-data-master/data/events/*'

# From Football Data Science by Bryan McGuire. Free to use with credit.
# https://www.footballdatascience.co.uk/research/do-substitutions-work
import json
import os
import random
from collections import Counter, defaultdict

DATA = os.path.expanduser("~/statsbomb-open-data/data")
WINDOW = 15  # minutes either side of the moment

# Whole seasons and whole tournaments only: skip sets where one team plays in more than a quarter of the matches
games = []
for comp in json.load(open(f"{DATA}/competitions.json")):
    matches = json.load(open(f"{DATA}/matches/{comp['competition_id']}/{comp['season_id']}.json"))
    seen = Counter(t for m in matches for t in (m["home_team"]["home_team_name"], m["away_team"]["away_team_name"]))
    if seen.most_common(1)[0][1] > len(matches) / 4:
        continue
    for m in matches:
        goals, shots, subs = [], [], []
        for e in json.load(open(f"{DATA}/events/{m['match_id']}.json")):
            if e["period"] > 2:  # normal time only
                continue
            kind, team, minute = e["type"]["name"], e["team"]["name"], e["minute"] + e["second"] / 60
            if kind == "Shot" and e["shot"]["outcome"]["name"] == "Goal" or kind == "Own Goal For":
                goals.append((e["period"], minute, team))
            if kind == "Shot" and e["period"] == 2 and e["shot"]["type"]["name"] != "Penalty":
                shots.append((minute, team, e["shot"]["statsbomb_xg"]))
            if kind == "Substitution" and e["period"] == 2 and e["substitution"].get("outcome", {}).get("name") == "Tactical":
                subs.append((minute, team))
        games.append({"teams": (m["home_team"]["home_team_name"], m["away_team"]["away_team_name"]),
                      "goals": goals, "shots": shots, "subs": subs, "five_subs": comp["season_name"] >= "2020", "women": comp["competition_gender"] == "female"})
print(f"{len(games)} matches: {sum(not g['women'] for g in games)} men's, {sum(g['women'] for g in games)} women's")

def margin(g, team, minute):  # goals for minus goals against before this minute of the second half
    return sum(1 if t == team else -1 for p, mn, t in g["goals"] if p == 1 or mn < minute)

def xg(g, team, a, b):  # non-penalty xG between minutes a and b of the second half
    return sum(x for mn, t, x in g["shots"] if t == team and a <= mn < b)

def moment(i, g, team, rival, t):  # what the 15 minutes either side of minute t looked like
    return {"match": i, "five_subs": g["five_subs"],
            "situation": (int(t), max(margin(g, team, t), -2),  # the minute, and one goal down or two or more
                          any(r == rival and p == 2 and t - WINDOW <= mn < t for p, mn, r in g["goals"])),  # just conceded?
            "for before": xg(g, team, t - WINDOW, t), "for after": xg(g, team, t, t + WINDOW),
            "against before": xg(g, rival, t - WINDOW, t), "against after": xg(g, rival, t, t + WINDOW),
            "scored again": any(r == team and p == 2 and mn >= t for p, mn, r in g["goals"]),
            "drew or won": margin(g, team, 999) >= 0}

changed, waited = [], []
for i, g in enumerate(games):
    for team, rival in (g["teams"], g["teams"][::-1]):
        subs = [mn for mn, t in g["subs"] if t == team]
        # the team's first change while behind, 60th to 80th minute, with no change in the 15 minutes before it
        first = next((mn for mn in subs if 60 <= mn < 80 and margin(g, team, mn) < 0), None)
        if first is not None and not any(first - WINDOW <= mn < first for mn in subs):
            changed.append(moment(i, g, team, rival, first))
        # every minute this team was behind and hadn't changed in the last 15 minutes or this minute, whatever it did later
        for t in range(60, 80):
            if margin(g, team, t) < 0 and not any(t - WINDOW <= mn < t + 1 for mn in subs):
                waited.append(moment(i, g, team, rival, t))

def per90(v, field):
    return v * 90 / WINDOW if field.startswith(("for", "against")) else v

def compare(ch, wt, field, match_on=3):  # each change against teams in the same situation that hadn't changed yet
    same = defaultdict(list)
    for w in wt:
        same[w["situation"][:match_on]].append(w[field])
    pairs = [(c[field], sum(s) / len(s)) for c in ch if (s := same.get(c["situation"][:match_on]))]
    return (per90(sum(a for a, _ in pairs) / len(pairs), field), per90(sum(b for _, b in pairs) / len(pairs), field), len(pairs))

rnd = random.Random(2026)  # fixed, so the luck margins repeat exactly
changed_in, waited_in = defaultdict(list), defaultdict(list)
for c in changed:
    changed_in[c["match"]].append(c)
for w in waited:
    waited_in[w["match"]].append(w)

def luck_margin(field, before=None, n=500):  # 95% range from resampling whole matches
    diffs = []
    for _ in range(n):
        pick = [rnd.randrange(len(games)) for _ in games]
        ch = [c for i in pick for c in changed_in[i]]
        wt = [w for i in pick for w in waited_in[i]]
        a, b, _ = compare(ch, wt, field)
        if before:
            a0, b0, _ = compare(ch, wt, before)
            a, b = a - a0, b - b0
        diffs.append(a - b)
    diffs.sort()
    return diffs[int(0.025 * n)], diffs[int(0.975 * n)]

print(f"first changes while behind, 60th to 80th minute: {len(changed)}; "
      f"moments behind without a change: {len(waited)}, in {len({w['match'] for w in waited})} matches")

# 1. Same minute, same margin
print("\nmatched on the minute and the margin. Non-penalty xG per 90, 15 minutes before -> 15 minutes after")
for side in ("for", "against"):
    (cb, wb, n), (ca, wa, _) = compare(changed, waited, f"{side} before", 2), compare(changed, waited, f"{side} after", 2)
    print(f"  {side:8} changed {cb:.2f} -> {ca:.2f} ({ca / cb - 1:+.0%}); waited {wb:.2f} -> {wa:.2f} ({wa / wb - 1:+.0%})")
same = defaultdict(list)
for x in waited:
    same[x["situation"][:2]].append(x["situation"][2])
just = [(x["situation"][2], sum(s) / len(s)) for x in changed if (s := same.get(x["situation"][:2]))]
print(f"  had conceded in the 15 minutes before: changed {sum(a for a, _ in just) / len(just):.0%}, "
      f"waited {sum(b for _, b in just) / len(just):.0%}")

# 2. Like for like: same minute, same margin, and whether they had just conceded
print("\nmatched on the minute, the margin and whether they had just conceded")
for side in ("for", "against"):
    (cb, wb, n), (ca, wa, _) = compare(changed, waited, f"{side} before"), compare(changed, waited, f"{side} after")
    lo, hi = luck_margin(f"{side} after", f"{side} before")
    print(f"  xG {side:8} changed {cb:.2f} -> {ca:.2f}; waited {wb:.2f} -> {wa:.2f}; "
          f"extra change {(ca - cb) - (wa - wb):+.2f} per 90 ({lo:+.2f} to {hi:+.2f}), {n} changes")
for field in ("scored again", "drew or won"):
    c, w, _ = compare(changed, waited, field)
    lo, hi = luck_margin(field)
    print(f"  {field}: changed {c:.1%}, waited {w:.1%}, difference {c - w:+.1%} ({lo:+.1%} to {hi:+.1%})")
for label, era in (("three-sub era", False), ("five-sub era", True)):
    ch, wt = [c for c in changed if c["five_subs"] == era], [w for w in waited if w["five_subs"] == era]
    (cb, wb, n), (ca, wa, _) = compare(ch, wt, "for before"), compare(ch, wt, "for after")
    print(f"  {label}: {n} changes, extra change in xG for {(ca - cb) - (wa - wb):+.2f} per 90")