Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 133 additions & 0 deletions app/analytics/player_hitting.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
"""Season-level hitting rates for one player.

Answers one question:

What did this player's overall offensive season look like?

The input is one stored season aggregate, not a list of games, so this is a
single calculation rather than a trend: no rolling windows, no chart points,
and no comparison with MLB. For an in-progress season the aggregate reflects
the most recent import, and nothing here implies the season is complete. A
player who played for more than one club is described by the combined season
aggregate; the individual team stints are not modelled.

Every rate is a ratio of season totals. A rate whose denominator is zero is
undefined and returned as ``None``, never ``0.0``: a player with no at-bats did
not bat ``.000``, and presenting it that way would fabricate a statistic.

Like the rest of the package this layer is free of FastAPI, Jinja, SQLAlchemy,
Plotly, and the MLB API. Values keep full precision; rounding is presentation.
"""

from app.schemas.analytics import PlayerHittingOverview, PlayerPlateAppearanceRates
from app.schemas.players import PlayerSeasonHitting


class PlayerHittingAnalysisError(ValueError):
"""A stored hitting line contradicts itself and cannot be described."""


def build_player_hitting_overview(
hitting: PlayerSeasonHitting,
) -> PlayerHittingOverview:
"""Derive AVG, OBP, SLG, OPS, and the plate-appearance rate profile.

Formulas, all over season totals:

- ``TB = H + 2B + 2 * 3B + 3 * HR``. ``H`` already counts the first base
of every hit, so each extra-base hit adds only its extra bases.
- ``AVG = H / AB``
- ``OBP = (H + BB + HBP) / (AB + BB + HBP + SF)``. Sacrifice bunts are not
in the denominator.
- ``SLG = TB / AB``
- ``OPS = OBP + SLG``, undefined when either component is.
- ``K% = SO / PA``, ``BB% = BB / PA``, ``HR% = HR / PA``.

Raises
------
PlayerHittingAnalysisError
The line records more of an outcome than the plate appearances or
at-bats that contain it, which no real season can.
"""
_require_subset_counts(hitting)

total_bases = (
hitting.hits + hitting.doubles + 2 * hitting.triples + 3 * hitting.home_runs
)
batting_average = _ratio(hitting.hits, hitting.at_bats)
on_base_percentage = _ratio(
hitting.hits + hitting.base_on_balls + hitting.hit_by_pitch,
hitting.at_bats
+ hitting.base_on_balls
+ hitting.hit_by_pitch
+ hitting.sac_flies,
)
slugging_percentage = _ratio(total_bases, hitting.at_bats)
on_base_plus_slugging = (
None
if on_base_percentage is None or slugging_percentage is None
else on_base_percentage + slugging_percentage
)

return PlayerHittingOverview(
hitting=hitting,
total_bases=total_bases,
batting_average=batting_average,
on_base_percentage=on_base_percentage,
slugging_percentage=slugging_percentage,
on_base_plus_slugging=on_base_plus_slugging,
plate_appearance_rates=_plate_appearance_rates(hitting),
)


def _require_subset_counts(hitting: PlayerSeasonHitting) -> None:
"""Reject a line whose numerators exceed the totals that contain them.

``PlayerSeasonHitting`` does not prove these relationships, and each one
bounds a rate derived here. Checking them explicitly turns a corrupted
stored row into a named data-integrity error rather than a schema failure
while constructing the analysis. A batting strikeout always ends an
at-bat, so it is bounded by at-bats as well as plate appearances.

Plate-appearance bounds are checked first so the error names the rate
denominator that was actually exceeded.
"""
plate_appearances = hitting.plate_appearances
for count, count_name, total, total_name in (
(hitting.strikeouts, "strikeouts", plate_appearances, "plate appearances"),
(hitting.base_on_balls, "walks", plate_appearances, "plate appearances"),
(hitting.home_runs, "home runs", plate_appearances, "plate appearances"),
(hitting.hits, "hits", hitting.at_bats, "at-bats"),
(hitting.strikeouts, "strikeouts", hitting.at_bats, "at-bats"),
):
if count > total:
raise PlayerHittingAnalysisError(
f"Player {hitting.player_id}'s stored {hitting.season} hitting "
f"line records {count} {count_name} in {total} {total_name}; "
f"{count_name} cannot exceed {total_name}"
)


def _plate_appearance_rates(
hitting: PlayerSeasonHitting,
) -> PlayerPlateAppearanceRates | None:
"""Return K%, BB%, and HR% together, or None with no plate appearances."""
plate_appearances = hitting.plate_appearances
if plate_appearances == 0:
return None
return PlayerPlateAppearanceRates(
plate_appearances=plate_appearances,
strikeouts=hitting.strikeouts,
base_on_balls=hitting.base_on_balls,
home_runs=hitting.home_runs,
strikeout_rate=hitting.strikeouts / plate_appearances,
walk_rate=hitting.base_on_balls / plate_appearances,
home_run_rate=hitting.home_runs / plate_appearances,
)


def _ratio(numerator: int, denominator: int) -> float | None:
"""Divide, or return None when the rate is undefined."""
if denominator == 0:
return None
return numerator / denominator
40 changes: 40 additions & 0 deletions app/database/repositories.py
Original file line number Diff line number Diff line change
Expand Up @@ -528,6 +528,46 @@ def list_player_catalog_seasons(session: Session) -> list[int]:
) from exc


def get_player_catalog_entry(
session: Session,
*,
player_id: int,
season: int,
) -> PlayerSeasonCatalogEntry | None:
"""Return one player's catalog membership for one season, or None.

Membership comes from ``player_seasons`` alone. A stored identity or a
stored hitting row does not by itself place a player in a season.
"""
stmt = (
select(PlayerSeasonCatalogRecord, PlayerRecord)
.join(
PlayerRecord,
PlayerRecord.player_id == PlayerSeasonCatalogRecord.player_id,
)
.where(
PlayerSeasonCatalogRecord.player_id == player_id,
PlayerSeasonCatalogRecord.season == season,
)
)
try:
row = session.execute(stmt).one_or_none()
except OperationalError as exc:
message = str(exc.orig if exc.orig is not None else exc).lower()
if "no such table" not in message or not (
"player_seasons" in message or "players" in message
):
raise
raise DatabaseSchemaMissingError(
"Player catalog tables are missing. "
f"Apply migrations with: {MIGRATION_HINT}"
) from exc
if row is None:
return None
membership, player = row
return membership.to_domain(player.to_domain())


def list_player_catalog(
session: Session, *, season: int
) -> list[PlayerSeasonCatalogEntry]:
Expand Down
175 changes: 174 additions & 1 deletion app/schemas/analytics.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Schemas for calculated team hitting analytics.
"""Schemas for calculated team and player analytics.

These models are the contract between the analytics layer and everything that
presents it. They carry finished numbers, not raw MLB payloads, and they keep
Expand All @@ -8,6 +8,9 @@
than through a shared metric type. They are read the same way but mean
different things, and honest duplication is cheaper to follow than an
abstraction covering four cases.

Player models describe one stored season aggregate rather than a game-by-game
trend, so they carry no chart points, rolling windows, or league context.
"""

from __future__ import annotations
Expand All @@ -18,6 +21,7 @@
from pydantic import BaseModel, ConfigDict, Field, model_validator

from app.schemas.games import HomeAway
from app.schemas.players import PlayerSeasonHitting


def _validate_prior_window_pair(
Expand Down Expand Up @@ -1676,3 +1680,172 @@ def _comparison_is_internally_consistent(self) -> TeamHitsAllowedLeagueCompariso
f"team_hits_allowed_per_game - league.hits_per_game ({expected})"
)
return self


def _validate_optional_ratio(
name: str,
value: float | None,
*,
numerator: int,
denominator: int,
) -> None:
"""Require a rate to equal its components, and be None only at zero.

An undefined rate and a real ``.000`` rate mean different things, so a zero
denominator must produce None and a non-zero one must produce the ratio.
"""
if denominator == 0:
if value is not None:
raise ValueError(f"{name} must be None when its denominator is zero")
return
expected = numerator / denominator
if value is None or not isclose(value, expected, rel_tol=1e-9, abs_tol=1e-9):
raise ValueError(
f"{name} ({value}) must equal {numerator} / {denominator} ({expected})"
)


class PlayerPlateAppearanceRates(BaseModel):
"""Share of a player's plate appearances ending in three outcomes.

All three rates share ``plate_appearances`` as their denominator, which is
why they are grouped: they describe how the player's stored season
outcomes were composed, not how good those outcomes were. The model only
exists when the player had at least one plate appearance.
"""

model_config = ConfigDict(frozen=True, extra="forbid")

plate_appearances: int = Field(gt=0, description="The shared denominator.")
strikeouts: int = Field(ge=0, description="Batting strikeouts.")
base_on_balls: int = Field(ge=0, description="Walks, including intentional.")
home_runs: int = Field(ge=0, description="Home runs.")
strikeout_rate: float = Field(
ge=0, le=1, description="K%: strikeouts / plate_appearances."
)
walk_rate: float = Field(
ge=0, le=1, description="BB%: base_on_balls / plate_appearances."
)
home_run_rate: float = Field(
ge=0, le=1, description="HR%: home_runs / plate_appearances."
)

@model_validator(mode="after")
def _rates_match_their_components(self) -> PlayerPlateAppearanceRates:
for name, numerator in (
("strikeout_rate", self.strikeouts),
("walk_rate", self.base_on_balls),
("home_run_rate", self.home_runs),
):
_validate_optional_ratio(
name,
getattr(self, name),
numerator=numerator,
denominator=self.plate_appearances,
)
return self


class PlayerHittingOverview(BaseModel):
"""One player's stored season hitting aggregate with its derived rates.

The stored season aggregate is carried unchanged beside the rates derived
from it, and the validators below prove the two agree. Each rate is None
exactly when its denominator is zero, never ``0.0``. OPS is None whenever
either OBP or SLG is.

This is the season aggregate as currently stored, which for an in-progress
season is the total at the most recent import rather than a completed
season. It is not a trend, does not compare the player with MLB, and says
nothing about which club or clubs the player played for.
"""

model_config = ConfigDict(frozen=True, extra="forbid")

hitting: PlayerSeasonHitting
total_bases: int = Field(ge=0, description="H + 2B + 2 * 3B + 3 * HR.")
batting_average: float | None = Field(
ge=0, le=1, description="AVG: hits / at_bats, or None with no at-bats."
)
on_base_percentage: float | None = Field(
ge=0,
le=1,
description="OBP: (H + BB + HBP) / (AB + BB + HBP + SF), or None when "
"that denominator is zero.",
)
slugging_percentage: float | None = Field(
ge=0, le=4, description="SLG: total_bases / at_bats, or None with no at-bats."
)
on_base_plus_slugging: float | None = Field(
ge=0, le=5, description="OPS: OBP + SLG, or None when either is None."
)
plate_appearance_rates: PlayerPlateAppearanceRates | None = Field(
description="K%, BB%, and HR%, or None with no plate appearances."
)

@model_validator(mode="after")
def _rates_match_the_stored_line(self) -> PlayerHittingOverview:
line = self.hitting
expected_total_bases = (
line.hits + line.doubles + 2 * line.triples + 3 * line.home_runs
)
if self.total_bases != expected_total_bases:
raise ValueError(
f"total_bases ({self.total_bases}) must equal H + 2B + 2 * 3B + "
f"3 * HR ({expected_total_bases})"
)
_validate_optional_ratio(
"batting_average",
self.batting_average,
numerator=line.hits,
denominator=line.at_bats,
)
_validate_optional_ratio(
"on_base_percentage",
self.on_base_percentage,
numerator=line.hits + line.base_on_balls + line.hit_by_pitch,
denominator=(
line.at_bats + line.base_on_balls + line.hit_by_pitch + line.sac_flies
),
)
_validate_optional_ratio(
"slugging_percentage",
self.slugging_percentage,
numerator=self.total_bases,
denominator=line.at_bats,
)
if self.on_base_percentage is None or self.slugging_percentage is None:
if self.on_base_plus_slugging is not None:
raise ValueError(
"on_base_plus_slugging must be None when OBP or SLG is None"
)
else:
expected_ops = self.on_base_percentage + self.slugging_percentage
if self.on_base_plus_slugging is None or not isclose(
self.on_base_plus_slugging, expected_ops, rel_tol=1e-9, abs_tol=1e-9
):
raise ValueError(
f"on_base_plus_slugging ({self.on_base_plus_slugging}) must "
f"equal OBP + SLG ({expected_ops})"
)
rates = self.plate_appearance_rates
if (rates is None) != (line.plate_appearances == 0):
raise ValueError(
"plate_appearance_rates must be present exactly when "
"plate_appearances is non-zero"
)
if rates is not None and (
rates.plate_appearances,
rates.strikeouts,
rates.base_on_balls,
rates.home_runs,
) != (
line.plate_appearances,
line.strikeouts,
line.base_on_balls,
line.home_runs,
):
raise ValueError(
"plate_appearance_rates components must match the stored line"
)
return self
Loading
Loading