paper-dynasty-database/app/services/evolution_evaluator.py

"""Evolution evaluator service (WP-08 / WP-13).

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-02 (PlayerSeasonStats), WP-04 (EvolutionCardState), 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 duck-type stats object with the
    attributes required by compute_value_for_track:
      batter: pa, hits, doubles, triples, hr
      sp/rp:  outs, strikeouts

    Note: PlayerSeasonStats stores pitcher strikeouts as 'k'; this class
    exposes them as 'strikeouts' to satisfy the formula engine Protocol.
    """

    __slots__ = ("pa", "hits", "doubles", "triples", "hr", "outs", "strikeouts")

    def __init__(self, pa, hits, doubles, triples, hr, outs, strikeouts):
        self.pa = pa
        self.hits = hits
        self.doubles = doubles
        self.triples = triples
        self.hr = hr
        self.outs = outs
        self.strikeouts = strikeouts


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 models are available).
        _state_model: Override for EvolutionCardState (used in tests).
        _compute_value_fn: Override for formula_engine.compute_value_for_track
            (used in tests).
        _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),
        # PlayerSeasonStats stores pitcher Ks as 'k'; expose as 'strikeouts'
        # to satisfy the PitcherStats Protocol expected by the formula engine.
        strikeouts=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 = card_state.current_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(),
    }