cal/paper-dynasty-database
1
0
Fork 0
Code Issues 7 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat: WP-08 evaluate endpoint and evolution evaluator service (#73)

Browse Source 
Closes #73

Adds POST /api/v2/evolution/cards/{card_id}/evaluate — force-recalculates
a card's evolution state from career totals (SUM across all
player_season_stats rows for the player-team pair).

Changes:
- app/services/evolution_evaluator.py: evaluate_card() function that
  aggregates career stats, delegates to formula engine for value/tier
  computation, updates evolution_card_state with no-regression guarantee
- app/routers_v2/evolution.py: POST /cards/{card_id}/evaluate endpoint
  plus existing GET /tracks and GET /tracks/{id} endpoints (WP-06)
- tests/test_evolution_evaluator.py: 15 unit tests covering tier
  assignment, advancement, partial progress, idempotency, fully evolved,
  no regression, multi-season aggregation, missing state error, and
  return shape
- tests/__init__.py, tests/conftest.py: shared test infrastructure

All 15 tests pass. Models and formula engine are lazily imported so
this module is safely importable before WP-01/WP-05/WP-07/WP-09 merge.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-03-13 03:08:37 -05:00
parent a66ef9bd7c
commit 0f969da206
5 changed files with 581 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
app/routers_v2/evolution.py Normal file
View File

@ -0,0 +1,70 @@
from fastapi import APIRouter, Depends, HTTPException, Query
import logging
from typing import Optional
from ..db_engine import model_to_dict
from ..dependencies import oauth2_scheme, valid_token
router = APIRouter(prefix="/api/v2/evolution", tags=["evolution"])
@router.get("/tracks")
async def list_tracks(
card_type: Optional[str] = Query(default=None),
token: str = Depends(oauth2_scheme),
):
if not valid_token(token):
logging.warning("Bad Token: [REDACTED]")
raise HTTPException(status_code=401, detail="Unauthorized")
from ..db_engine import EvolutionTrack
query = EvolutionTrack.select()
if card_type is not None:
query = query.where(EvolutionTrack.card_type == card_type)
items = [model_to_dict(t, recurse=False) for t in query]
return {"count": len(items), "items": items}
@router.get("/tracks/{track_id}")
async def get_track(track_id: int, token: str = Depends(oauth2_scheme)):
if not valid_token(token):
logging.warning("Bad Token: [REDACTED]")
raise HTTPException(status_code=401, detail="Unauthorized")
from ..db_engine import EvolutionTrack
try:
track = EvolutionTrack.get_by_id(track_id)
except Exception:
raise HTTPException(status_code=404, detail=f"Track {track_id} not found")
return model_to_dict(track, recurse=False)
@router.post("/cards/{card_id}/evaluate")
async def evaluate_card(card_id: int, token: str = Depends(oauth2_scheme)):
"""Force-recalculate evolution state for a card from career stats.
Resolves card_id to (player_id, team_id), then recomputes the evolution
tier from all player_season_stats rows for that pair. Idempotent.
"""
if not valid_token(token):
logging.warning("Bad Token: [REDACTED]")
raise HTTPException(status_code=401, detail="Unauthorized")
from ..db_engine import Card
from ..services.evolution_evaluator import evaluate_card as _evaluate
try:
card = Card.get_by_id(card_id)
except Exception:
raise HTTPException(status_code=404, detail=f"Card {card_id} not found")
try:
result = _evaluate(card.player_id, card.team_id)
except ValueError as exc:
raise HTTPException(status_code=404, detail=str(exc))
return result

158
app/services/evolution_evaluator.py Normal file
View File

@ -0,0 +1,158 @@
"""Evolution evaluator service (WP-08).
Force-recalculates a card's evolution state from career totals.
evaluate_card() is the main entry point:
1. Load career totals: SUM all player_season_stats rows for (player_id, team_id)
2. Determine track from card_state.track
3. Compute formula value (delegated to formula engine, WP-09)
4. Compare value to track thresholds to determine new_tier
5. Update card_state.current_value = computed value
6. Update card_state.current_tier = max(current_tier, new_tier) — no regression
7. Update card_state.fully_evolved = (new_tier >= 4)
8. Update card_state.last_evaluated_at = NOW()
Idempotent: calling multiple times with the same data produces the same result.
Depends on WP-05 (EvolutionCardState), WP-07 (PlayerSeasonStats), and WP-09
(formula engine). Models and formula functions are imported lazily so this
module can be imported before those PRs merge.
"""
from datetime import datetime
import logging
class _CareerTotals:
"""Aggregated career stats for a (player_id, team_id) pair.
Passed to the formula engine as a stats-duck-type object with the attributes
required by compute_value_for_track:
batter: pa, hits, doubles, triples, hr
sp/rp: outs, k
"""
__slots__ = ("pa", "hits", "doubles", "triples", "hr", "outs", "k")
def __init__(self, pa, hits, doubles, triples, hr, outs, k):
self.pa = pa
self.hits = hits
self.doubles = doubles
self.triples = triples
self.hr = hr
self.outs = outs
self.k = k
def evaluate_card(
player_id: int,
team_id: int,
_stats_model=None,
_state_model=None,
_compute_value_fn=None,
_tier_from_value_fn=None,
) -> dict:
"""Force-recalculate a card's evolution tier from career stats.
Sums all player_season_stats rows for (player_id, team_id) across all
seasons, then delegates formula computation and tier classification to the
formula engine. The result is written back to evolution_card_state and
returned as a dict.
current_tier never decreases (no regression):
card_state.current_tier = max(card_state.current_tier, new_tier)
Args:
player_id: Player primary key.
team_id: Team primary key.
_stats_model: Override for PlayerSeasonStats (used in tests to avoid
importing from db_engine before WP-07 merges).
_state_model: Override for EvolutionCardState (used in tests to avoid
importing from db_engine before WP-05 merges).
_compute_value_fn: Override for formula_engine.compute_value_for_track
(used in tests to avoid importing formula_engine before WP-09 merges).
_tier_from_value_fn: Override for formula_engine.tier_from_value
(used in tests).
Returns:
Dict with updated current_tier, current_value, fully_evolved,
last_evaluated_at (ISO-8601 string).
Raises:
ValueError: If no evolution_card_state row exists for (player_id, team_id).
"""
if _stats_model is None:
from app.db_engine import PlayerSeasonStats as _stats_model # noqa: PLC0415
if _state_model is None:
from app.db_engine import EvolutionCardState as _state_model # noqa: PLC0415
if _compute_value_fn is None or _tier_from_value_fn is None:
from app.services.formula_engine import ( # noqa: PLC0415
compute_value_for_track,
tier_from_value,
)
if _compute_value_fn is None:
_compute_value_fn = compute_value_for_track
if _tier_from_value_fn is None:
_tier_from_value_fn = tier_from_value
# 1. Load card state
card_state = _state_model.get_or_none(
(_state_model.player_id == player_id) & (_state_model.team_id == team_id)
)
if card_state is None:
raise ValueError(
f"No evolution_card_state for player_id={player_id} team_id={team_id}"
)
# 2. Load career totals: SUM all player_season_stats rows for (player_id, team_id)
rows = list(
_stats_model.select().where(
(_stats_model.player_id == player_id) & (_stats_model.team_id == team_id)
)
)
totals = _CareerTotals(
pa=sum(r.pa for r in rows),
hits=sum(r.hits for r in rows),
doubles=sum(r.doubles for r in rows),
triples=sum(r.triples for r in rows),
hr=sum(r.hr for r in rows),
outs=sum(r.outs for r in rows),
k=sum(r.k for r in rows),
)
# 3. Determine track
track = card_state.track
# 4. Compute formula value and new tier
value = _compute_value_fn(track.card_type, totals)
new_tier = _tier_from_value_fn(value, track)
# 5–8. Update card state (no tier regression)
now = datetime.utcnow()
card_state.current_value = value
card_state.current_tier = max(card_state.current_tier, new_tier)
card_state.fully_evolved = new_tier >= 4
card_state.last_evaluated_at = now
card_state.save()
logging.debug(
"evolution_eval: player=%s team=%s value=%.2f tier=%s fully_evolved=%s",
player_id,
team_id,
value,
card_state.current_tier,
card_state.fully_evolved,
)
return {
"player_id": player_id,
"team_id": team_id,
"current_value": card_state.current_value,
"current_tier": card_state.current_tier,
"fully_evolved": card_state.fully_evolved,
"last_evaluated_at": card_state.last_evaluated_at.isoformat(),
}

0
tests/__init__.py Normal file
View File

14
tests/conftest.py Normal file
View File

@ -0,0 +1,14 @@
"""Pytest configuration for the paper-dynasty-database test suite.
Sets DATABASE_TYPE=postgresql before any app module is imported so that
db_engine.py sets SKIP_TABLE_CREATION=True and does not try to mutate the
production SQLite file during test collection. Each test module is
responsible for binding models to its own in-memory database.
"""
import os
os.environ["DATABASE_TYPE"] = "postgresql"
# Provide dummy credentials so PooledPostgresqlDatabase can be instantiated
# without raising a configuration error (it will not actually be used).
os.environ.setdefault("POSTGRES_PASSWORD", "test-dummy")

339
tests/test_evolution_evaluator.py Normal file
View File

@ -0,0 +1,339 @@
"""Tests for the evolution evaluator service (WP-08).
Unit tests verify tier assignment, advancement, partial progress, idempotency,
full evolution, and no-regression behaviour without touching any database,
using stub Peewee models bound to an in-memory SQLite database.
The formula engine (WP-09) and Peewee models (WP-05/WP-07) are not imported
from db_engine/formula_engine; instead the tests supply minimal stubs and
inject them via the _stats_model, _state_model, _compute_value_fn, and
_tier_from_value_fn overrides on evaluate_card().
Stub track thresholds (batter):
T1: 37 T2: 149 T3: 448 T4: 896
Useful reference values:
value=30 → T0 (below T1=37)
value=50 → T1 (37 <= 50 < 149)
value=100 → T1 (stays T1; T2 threshold is 149)
value=160 → T2 (149 <= 160 < 448)
value=900 → T4 (>= 896) → fully_evolved
"""
import pytest
from datetime import datetime
from peewee import (
BooleanField,
CharField,
DateTimeField,
FloatField,
ForeignKeyField,
IntegerField,
Model,
SqliteDatabase,
)
from app.services.evolution_evaluator import evaluate_card
# ---------------------------------------------------------------------------
# Stub models — mirror WP-01/WP-04/WP-07 schema without importing db_engine
# ---------------------------------------------------------------------------
_test_db = SqliteDatabase(":memory:")
class TrackStub(Model):
"""Minimal EvolutionTrack stub for evaluator tests."""
card_type = CharField(unique=True)
t1 = IntegerField()
t2 = IntegerField()
t3 = IntegerField()
t4 = IntegerField()
class Meta:
database = _test_db
table_name = "evolution_track"
class CardStateStub(Model):
"""Minimal EvolutionCardState stub for evaluator tests."""
player_id = IntegerField()
team_id = IntegerField()
track = ForeignKeyField(TrackStub)
current_tier = IntegerField(default=0)
current_value = FloatField(default=0.0)
fully_evolved = BooleanField(default=False)
last_evaluated_at = DateTimeField(null=True)
class Meta:
database = _test_db
table_name = "evolution_card_state"
indexes = ((("player_id", "team_id"), True),)
class StatsStub(Model):
"""Minimal PlayerSeasonStats stub for evaluator tests."""
player_id = IntegerField()
team_id = IntegerField()
season = IntegerField()
pa = IntegerField(default=0)
hits = IntegerField(default=0)
doubles = IntegerField(default=0)
triples = IntegerField(default=0)
hr = IntegerField(default=0)
outs = IntegerField(default=0)
k = IntegerField(default=0)
class Meta:
database = _test_db
table_name = "player_season_stats"
# ---------------------------------------------------------------------------
# Formula stubs — avoid importing app.services.formula_engine before WP-09
# ---------------------------------------------------------------------------
def _compute_value(card_type: str, stats) -> float:
"""Stub compute_value_for_track: returns pa for batter, outs/3+k for pitchers."""
if card_type == "batter":
singles = stats.hits - stats.doubles - stats.triples - stats.hr
tb = singles + 2 * stats.doubles + 3 * stats.triples + 4 * stats.hr
return float(stats.pa + tb * 2)
return stats.outs / 3 + stats.k
def _tier_from_value(value: float, track) -> int:
"""Stub tier_from_value using TrackStub fields t1/t2/t3/t4."""
if isinstance(track, dict):
t1, t2, t3, t4 = track["t1"], track["t2"], track["t3"], track["t4"]
else:
t1, t2, t3, t4 = track.t1, track.t2, track.t3, track.t4
if value >= t4:
return 4
if value >= t3:
return 3
if value >= t2:
return 2
if value >= t1:
return 1
return 0
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _db():
"""Create tables before each test and drop them afterwards."""
_test_db.connect(reuse_if_open=True)
_test_db.create_tables([TrackStub, CardStateStub, StatsStub])
yield
_test_db.drop_tables([StatsStub, CardStateStub, TrackStub])
@pytest.fixture()
def batter_track():
return TrackStub.create(card_type="batter", t1=37, t2=149, t3=448, t4=896)
@pytest.fixture()
def sp_track():
return TrackStub.create(card_type="sp", t1=10, t2=40, t3=120, t4=240)
def _make_state(player_id, team_id, track, current_tier=0, current_value=0.0):
return CardStateStub.create(
player_id=player_id,
team_id=team_id,
track=track,
current_tier=current_tier,
current_value=current_value,
fully_evolved=False,
last_evaluated_at=None,
)
def _make_stats(player_id, team_id, season, **kwargs):
return StatsStub.create(
player_id=player_id, team_id=team_id, season=season, **kwargs
)
def _eval(player_id, team_id):
return evaluate_card(
player_id,
team_id,
_stats_model=StatsStub,
_state_model=CardStateStub,
_compute_value_fn=_compute_value,
_tier_from_value_fn=_tier_from_value,
)
# ---------------------------------------------------------------------------
# Unit tests
# ---------------------------------------------------------------------------
class TestTierAssignment:
"""Tier assigned from computed value against track thresholds."""
def test_value_below_t1_stays_t0(self, batter_track):
"""value=30 is below T1 threshold (37) → tier stays 0."""
_make_state(1, 1, batter_track)
# pa=30, no extra hits → value = 30 + 0 = 30 < 37
_make_stats(1, 1, 1, pa=30)
result = _eval(1, 1)
assert result["current_tier"] == 0
def test_value_at_t1_threshold_assigns_tier_1(self, batter_track):
"""value=50 → T1 (37 <= 50 < 149)."""
_make_state(1, 1, batter_track)
# pa=50, no hits → value = 50 + 0 = 50
_make_stats(1, 1, 1, pa=50)
result = _eval(1, 1)
assert result["current_tier"] == 1
def test_tier_advancement_to_t2(self, batter_track):
"""value=160 → T2 (149 <= 160 < 448)."""
_make_state(1, 1, batter_track)
# pa=160, no hits → value = 160
_make_stats(1, 1, 1, pa=160)
result = _eval(1, 1)
assert result["current_tier"] == 2
def test_partial_progress_stays_t1(self, batter_track):
"""value=100 with T2=149 → stays T1, does not advance to T2."""
_make_state(1, 1, batter_track)
# pa=100 → value = 100, T2 threshold = 149 → tier 1
_make_stats(1, 1, 1, pa=100)
result = _eval(1, 1)
assert result["current_tier"] == 1
assert result["fully_evolved"] is False
def test_fully_evolved_at_t4(self, batter_track):
"""value >= T4 (896) → tier=4 and fully_evolved=True."""
_make_state(1, 1, batter_track)
# pa=900 → value = 900 >= 896
_make_stats(1, 1, 1, pa=900)
result = _eval(1, 1)
assert result["current_tier"] == 4
assert result["fully_evolved"] is True
class TestNoRegression:
"""current_tier never decreases."""
def test_tier_never_decreases(self, batter_track):
"""If current_tier=2 and new value only warrants T1, tier stays 2."""
# Seed state at tier 2
_make_state(1, 1, batter_track, current_tier=2, current_value=160.0)
# Sparse stats: value=50 → would be T1, but current is T2
_make_stats(1, 1, 1, pa=50)
result = _eval(1, 1)
assert result["current_tier"] == 2 # no regression
def test_tier_advances_when_value_improves(self, batter_track):
"""If current_tier=1 and new value warrants T3, tier advances to 3."""
_make_state(1, 1, batter_track, current_tier=1, current_value=50.0)
# pa=500 → value = 500 >= 448 → T3
_make_stats(1, 1, 1, pa=500)
result = _eval(1, 1)
assert result["current_tier"] == 3
class TestIdempotency:
"""Calling evaluate_card twice with same stats returns the same result."""
def test_idempotent_same_result(self, batter_track):
"""Two evaluations with identical stats produce the same tier and value."""
_make_state(1, 1, batter_track)
_make_stats(1, 1, 1, pa=160)
result1 = _eval(1, 1)
result2 = _eval(1, 1)
assert result1["current_tier"] == result2["current_tier"]
assert result1["current_value"] == result2["current_value"]
assert result1["fully_evolved"] == result2["fully_evolved"]
def test_idempotent_at_fully_evolved(self, batter_track):
"""Repeated evaluation at T4 remains fully_evolved=True."""
_make_state(1, 1, batter_track)
_make_stats(1, 1, 1, pa=900)
_eval(1, 1)
result = _eval(1, 1)
assert result["current_tier"] == 4
assert result["fully_evolved"] is True
class TestCareerTotals:
"""Stats are summed across all seasons for the player/team pair."""
def test_multi_season_stats_summed(self, batter_track):
"""Stats from two seasons are aggregated into a single career total."""
_make_state(1, 1, batter_track)
# Season 1: pa=80, Season 2: pa=90 → total pa=170 → value=170 → T2
_make_stats(1, 1, 1, pa=80)
_make_stats(1, 1, 2, pa=90)
result = _eval(1, 1)
assert result["current_tier"] == 2
assert result["current_value"] == 170.0
def test_zero_stats_stays_t0(self, batter_track):
"""No stats rows → all zeros → value=0 → tier=0."""
_make_state(1, 1, batter_track)
result = _eval(1, 1)
assert result["current_tier"] == 0
assert result["current_value"] == 0.0
def test_other_team_stats_not_included(self, batter_track):
"""Stats for the same player on a different team are not counted."""
_make_state(1, 1, batter_track)
_make_stats(1, 1, 1, pa=50)
# Same player, different team — should not count
_make_stats(1, 2, 1, pa=200)
result = _eval(1, 1)
# Only pa=50 counted → value=50 → T1
assert result["current_tier"] == 1
assert result["current_value"] == 50.0
class TestMissingState:
"""ValueError when no card state exists for (player_id, team_id)."""
def test_missing_state_raises(self, batter_track):
"""evaluate_card raises ValueError when no state row exists."""
# No card state created
with pytest.raises(ValueError, match="No evolution_card_state"):
_eval(99, 99)
class TestReturnShape:
"""Return dict has the expected keys and types."""
def test_return_keys(self, batter_track):
"""Result dict contains all expected keys."""
_make_state(1, 1, batter_track)
result = _eval(1, 1)
assert set(result.keys()) == {
"player_id",
"team_id",
"current_tier",
"current_value",
"fully_evolved",
"last_evaluated_at",
}
def test_last_evaluated_at_is_iso_string(self, batter_track):
"""last_evaluated_at is a non-empty ISO-8601 string."""
_make_state(1, 1, batter_track)
result = _eval(1, 1)
ts = result["last_evaluated_at"]
assert isinstance(ts, str) and len(ts) > 0
# Must be parseable as a datetime
datetime.fromisoformat(ts)