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

"""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 BattingSeasonStats/PitchingSeasonStats 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 (BattingSeasonStats/PitchingSeasonStats),
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, strikeouts
    """

    __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 BattingSeasonStats or PitchingSeasonStats rows (based on
    card_type) 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 BattingSeasonStats/PitchingSeasonStats
            (used in tests to inject a stub model with all stat fields).
        _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 _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 from the appropriate season stats table
    if _stats_model is not None:
        # Test override: use the injected stub model for all fields
        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),
            strikeouts=sum(r.strikeouts for r in rows),
        )
    else:
        from app.db_engine import (
            BattingSeasonStats,
            PitchingSeasonStats,
        )  # noqa: PLC0415

        card_type = card_state.track.card_type
        if card_type == "batter":
            rows = list(
                BattingSeasonStats.select().where(
                    (BattingSeasonStats.player == player_id)
                    & (BattingSeasonStats.team == 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=0,
                strikeouts=sum(r.strikeouts for r in rows),
            )
        else:
            rows = list(
                PitchingSeasonStats.select().where(
                    (PitchingSeasonStats.player == player_id)
                    & (PitchingSeasonStats.team == team_id)
                )
            )
            totals = _CareerTotals(
                pa=0,
                hits=0,
                doubles=0,
                triples=0,
                hr=0,
                outs=sum(r.outs for r in rows),
                strikeouts=sum(r.strikeouts 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.now()
    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(),
    }