Coverage for src/usls/application/_analytics_helpers.py: 89%
91 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-05 10:11 +0000
1"""Pure helpers for the schedule-strength analytics use cases.
3Extracted from `_helpers.py` so the general-purpose helpers stay small and
4the analytics math (PPG indexing, tier classification, opponent walking) lives
5in one cohesive place. All functions here are side-effect-free transformations
6over domain entities — no I/O, no port calls.
7"""
9from dataclasses import dataclass
11from ..domain.exceptions import USLSNotFoundError
12from ..domain.models import Match, MatchCompetitor, Standing, Team, TierRecord
15@dataclass(frozen=True)
16class _PPGEntry:
17 """Cached per-team standings summary used by the SoS analytics."""
19 matches_played: int
20 points: int
21 ppg: float
24def _build_ppg_index(standings: list[Standing]) -> dict[str, _PPGEntry]:
25 """Build a `team_id -> _PPGEntry` map from a standings table."""
26 index: dict[str, _PPGEntry] = {}
27 for s in standings:
28 mp = s.wins + s.losses + s.ties
29 ppg = s.points / mp if mp else 0.0
30 index[s.team.id] = _PPGEntry(matches_played=mp, points=s.points, ppg=ppg)
31 return index
34def _find_team_in_standings(standings: list[Standing], team_id: str) -> Team | None:
35 """Return the Team for a given id from a standings table, or None if absent."""
36 return next((s.team for s in standings if s.team.id == team_id), None)
39def _team_from_schedule(schedule: list[Match], team_id: str) -> Team:
40 """Recover a Team object from a schedule when standings don't contain it.
42 Raises:
43 USLSNotFoundError: If neither standings nor schedule reference the team.
44 """
45 for match in schedule:
46 for comp in match.competitors:
47 if comp.team.id == team_id:
48 return comp.team
49 raise USLSNotFoundError(f"Team not found: {team_id}")
52def _resolve_team(standings: list[Standing], schedule: list[Match], team_id: str) -> Team:
53 """Return the Team from standings, falling back to the schedule if absent.
55 Raises:
56 USLSNotFoundError: If neither standings nor schedule reference the team.
57 """
58 found = _find_team_in_standings(standings, team_id)
59 return found if found is not None else _team_from_schedule(schedule, team_id)
62def _split_competitors(match: Match, team_id: str) -> tuple[MatchCompetitor | None, MatchCompetitor | None]:
63 """Return (our_side, opponent_side) for a match, or (None, None) if team not present."""
64 ours = next((c for c in match.competitors if c.team.id == team_id), None)
65 theirs = next((c for c in match.competitors if c.team.id != team_id), None)
66 if ours is None or theirs is None:
67 return None, None
68 return ours, theirs
71def _played_opponents(schedule: list[Match], team_id: str) -> list[Team]:
72 """Return the opponent Team for each completed match in `schedule`.
74 Skips unfinished matches (status_type != 'post') and any match that doesn't
75 contain the requested team. Multiple meetings against the same opponent
76 appear multiple times, so the caller can aggregate.
77 """
78 opponents: list[Team] = []
79 for match in schedule:
80 if match.status_type != "post":
81 continue
82 _ours, theirs = _split_competitors(match, team_id)
83 if theirs is not None:
84 opponents.append(theirs.team)
85 return opponents
88def _opponent_ppgs(schedule: list[Match], team_id: str, ppg_index: dict[str, _PPGEntry]) -> list[_PPGEntry]:
89 """Return the PPG entry for each opponent the team has played, skipping unknowns."""
90 return [ppg_index[o.id] for o in _played_opponents(schedule, team_id) if o.id in ppg_index]
93def _record_result(counters: list[int], ours: MatchCompetitor, theirs: MatchCompetitor) -> None:
94 """Increment win/loss/tie counters in-place from one match outcome.
96 counters is mutated as `[wins, losses, ties]`. A draw is recorded when
97 neither side has `winner=True`.
98 """
99 if ours.winner:
100 counters[0] += 1
101 elif theirs.winner:
102 counters[1] += 1
103 else:
104 counters[2] += 1
107def _classify_tier(rank: int, tier_specs: tuple[tuple[str, int, int], ...]) -> str:
108 """Return the tier name (e.g. 'Top', 'Middle', 'Bottom') for a given rank.
110 Tiers are evaluated in order; the first matching `low <= rank <= high` wins.
112 Raises:
113 ValueError: If rank doesn't fall within any spec — indicates a bug
114 in the caller (tier_specs should cover every possible rank).
115 """
116 for name, low, high in tier_specs:
117 if low <= rank <= high:
118 return name
119 raise ValueError(f"rank {rank} does not fall within any tier in {tier_specs}")
122def _tally_tier_results(
123 schedule: list[Match],
124 team_id: str,
125 rank_by_id: dict[str, int],
126 tier_specs: tuple[tuple[str, int, int], ...],
127) -> dict[str, list[int]]:
128 """Walk completed matches and tally W-L-T per tier name."""
129 tally: dict[str, list[int]] = {name: [0, 0, 0] for name, _, _ in tier_specs}
130 for match in schedule:
131 if match.status_type != "post":
132 continue
133 ours, theirs = _split_competitors(match, team_id)
134 if ours is None or theirs is None:
135 continue
136 rank = rank_by_id.get(theirs.team.id)
137 if rank is None:
138 continue
139 _record_result(tally[_classify_tier(rank, tier_specs)], ours, theirs)
140 return tally
143def _league_average_ppg(standings: list[Standing]) -> float:
144 """Compute league-wide average PPG (total points / total matches played)."""
145 total_points = sum(s.points for s in standings)
146 total_matches = sum(s.wins + s.losses + s.ties for s in standings)
147 return total_points / total_matches if total_matches else 0.0
150def _self_record(ppg_index: dict[str, _PPGEntry], team_id: str) -> tuple[int, int, float]:
151 """Return `(matches_played, points, raw_ppg)` for the given team, or zeros if absent.
153 Returns zeros when a team has been resolved (via `_resolve_team`) but is not
154 yet in the standings table — this happens for expansion teams that appear
155 in fixture data before they've played a match. In practice the caller has
156 already established the team exists; the zeros are a safe default for the
157 "no record yet" case.
158 """
159 entry = ppg_index.get(team_id)
160 if entry is None:
161 return 0, 0, 0.0
162 return entry.matches_played, entry.points, entry.ppg
165def _safe_ratio(numerator: float, denominator: float) -> float:
166 """Return numerator/denominator, or 0.0 if denominator is zero."""
167 return numerator / denominator if denominator else 0.0
170def _mean(values: list[float]) -> float:
171 """Return the arithmetic mean of `values`, or 0.0 if empty."""
172 return sum(values) / len(values) if values else 0.0
175def _validate_team_id(team_id: str) -> str:
176 """Return a stripped team_id, or raise ValueError if empty/whitespace."""
177 if not team_id or not team_id.strip():
178 raise ValueError("team_id must not be empty")
179 return team_id.strip()
182def _validate_tier_size(tier_size: int, league_size: int) -> None:
183 """Raise ValueError if tier_size is outside [1, league_size // 2]."""
184 if tier_size < 1 or 2 * tier_size > league_size:
185 raise ValueError(f"tier_size must be between 1 and {league_size // 2} for a {league_size}-team league")
188def _build_tier_specs(tier_size: int, league_size: int) -> tuple[tuple[str, int, int], ...]:
189 """Return (name, rank_low, rank_high) specs for top/middle/bottom tiers."""
190 return (
191 ("Top", 1, tier_size),
192 ("Middle", tier_size + 1, league_size - tier_size),
193 ("Bottom", league_size - tier_size + 1, league_size),
194 )
197def _build_tier_record(name: str, low: int, high: int, tally: dict[str, list[int]]) -> TierRecord:
198 """Build a TierRecord row from a tier name and its tallied counters."""
199 return TierRecord(
200 label=f"{name} {high - low + 1}",
201 rank_low=low,
202 rank_high=high,
203 wins=tally[name][0],
204 losses=tally[name][1],
205 ties=tally[name][2],
206 )