80e94498fa
Implements defensive error handling for WPA (Win Probability Added) calculations when rare game states are missing from the static lookup table. The safe_wpa_lookup() function uses a three-tier fallback strategy: exact lookup, bases-empty fallback, and 0.0 default value. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
472 lines
15 KiB
Python
472 lines
15 KiB
Python
import pytest
|
|
from sqlmodel import Session, select, func
|
|
|
|
from command_logic.logic_gameplay import advance_runners, doubles, gb_result_1, get_obc, get_re24, get_wpa, complete_play, log_run_scored, strikeouts, steals, xchecks, walks, popouts, hit_by_pitch, homeruns, singles, triples, bunts, chaos, safe_wpa_lookup
|
|
from in_game.gameplay_models import Lineup, Play
|
|
from tests.factory import session_fixture, Game
|
|
|
|
|
|
def test_advance_runners(session: Session):
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
assert play_1.starting_outs == 0
|
|
assert play_1.on_first_id is None
|
|
assert play_1.on_second_id is None
|
|
assert play_1.on_third_id is None
|
|
|
|
play_1.pa, play_1.ab, play_1.hit, play_1.double, play_1.batter_final = 1, 1, 1, 1, 2
|
|
|
|
play_2 = complete_play(session, play_1)
|
|
assert play_2.inning_half == play_1.inning_half
|
|
assert play_2.on_second_id == play_1.batter_id
|
|
|
|
play_2.pa, play_2.ab, play_2.hit, play_2.batter_final = 1, 1, 1, 1
|
|
play_2 = advance_runners(session, play_2, 1)
|
|
session.add(play_2)
|
|
session.commit()
|
|
|
|
assert play_2.on_second_final == 3
|
|
assert play_2.batter_final == 1
|
|
|
|
play_3 = complete_play(session, play_2)
|
|
|
|
assert play_3.on_third is not None
|
|
assert play_3.on_second is None
|
|
assert play_3.on_first is not None
|
|
|
|
play_3.pa, play_3.ab, play_3.hit, play_3.batter_final = 1, 1, 1, 2
|
|
play_3 = advance_runners(session, play_3, 3, earned_bases=1)
|
|
session.add(play_3)
|
|
session.commit()
|
|
|
|
assert play_3.rbi == 1
|
|
assert play_2.run == 1
|
|
assert play_2.e_run == 0
|
|
assert play_1.run == 1
|
|
assert play_1.e_run == 1
|
|
|
|
|
|
def test_get_obc():
|
|
assert get_obc() == 0
|
|
assert get_obc(on_first=True) == 1
|
|
assert get_obc(on_second=True) == 2
|
|
assert get_obc(on_third=True) == 3
|
|
assert get_obc(on_first=True, on_second=True) == 4
|
|
assert get_obc(on_first=True, on_third=True) == 5
|
|
assert get_obc(on_second=True, on_third=True) == 6
|
|
assert get_obc(on_first=True, on_second=True, on_third=True) == 7
|
|
|
|
|
|
def test_get_re24(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
this_play = game_1.current_play_or_none(session)
|
|
|
|
assert this_play is not None
|
|
assert get_re24(this_play, runs_scored=0, new_obc=0, new_starting_outs=2) == -.154
|
|
assert get_re24(this_play, runs_scored=1, new_obc=0, new_starting_outs=1) == 1
|
|
assert get_re24(this_play, runs_scored=0, new_obc=2, new_starting_outs=1) == 0.365
|
|
assert get_re24(this_play, runs_scored=0, new_obc=6, new_starting_outs=2) == 0.217
|
|
|
|
|
|
def test_get_wpa(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
next_play = game_1.current_play_or_none(session) # Starting wpa: 0.564
|
|
this_play = game_1.plays[0] # Starting wpa: 0.500
|
|
|
|
assert this_play != next_play
|
|
assert get_wpa(this_play, next_play) == -0.064
|
|
|
|
next_play.starting_outs = 2
|
|
next_play.away_score = 3 # Starting wpa: 0.347
|
|
|
|
assert get_wpa(this_play, next_play) == 0.153
|
|
|
|
|
|
def test_safe_wpa_lookup():
|
|
"""Test safe_wpa_lookup handles both valid and missing keys gracefully."""
|
|
# Test 1: Valid key should return actual value
|
|
valid_result = safe_wpa_lookup('bot', 1, 0, 0, 0)
|
|
assert isinstance(valid_result, float)
|
|
assert 0.0 <= valid_result <= 1.0
|
|
assert valid_result == 0.589 # Known value from wpa_data.csv
|
|
|
|
# Test 2: Missing key with bases loaded should fallback to bases empty
|
|
# This simulates the error case: top_1_0_out_6_obc_-5_home_run_diff doesn't exist
|
|
fallback_result = safe_wpa_lookup('top', 1, 0, 6, -5)
|
|
assert isinstance(fallback_result, float)
|
|
assert 0.0 <= fallback_result <= 1.0
|
|
# Should use fallback to bases empty state if available
|
|
|
|
# Test 3: If even bases empty fallback doesn't exist, should return 0.0
|
|
# (This is an edge case that shouldn't happen with complete data)
|
|
extreme_fallback = safe_wpa_lookup('top', 1, 0, 7, -10) # Invalid obc and extreme run diff
|
|
assert extreme_fallback == 0.0
|
|
|
|
# Test 4: Another valid key
|
|
valid_result_2 = safe_wpa_lookup('top', 9, 2, 3, 1)
|
|
assert isinstance(valid_result_2, float)
|
|
assert 0.0 <= valid_result_2 <= 1.0
|
|
|
|
|
|
def test_complete_play(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
this_play = game_1.current_play_or_none(session)
|
|
|
|
assert this_play.inning_half == 'top'
|
|
assert this_play.inning_num == 1
|
|
assert this_play.starting_outs == 1
|
|
|
|
this_play.pa, this_play.ab, this_play.so, this_play.outs = 1, 1, 1, 1
|
|
next_play = complete_play(session, this_play)
|
|
|
|
assert next_play.inning_half == 'top'
|
|
assert this_play.inning_num == 1
|
|
assert next_play.starting_outs == 2
|
|
assert next_play.is_tied
|
|
assert next_play.managerai == this_play.managerai
|
|
assert this_play.re24 == -0.154
|
|
|
|
next_play.pa, next_play.ab, next_play.double, next_play.batter_final = 1, 1, 1, 2
|
|
third_play = complete_play(session, next_play)
|
|
|
|
assert third_play.on_base_code == 2
|
|
assert next_play.re24 == 0.182
|
|
assert third_play.on_second == next_play.batter
|
|
|
|
third_play.pa, third_play.ab, third_play.hit
|
|
|
|
|
|
def test_complete_play_scratch(session: Session):
|
|
this_game = session.get(Game, 3)
|
|
|
|
assert len(this_game.plays) == 0
|
|
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
assert play_1.starting_outs == 0
|
|
assert play_1.on_first_id is None
|
|
assert play_1.on_second_id is None
|
|
assert play_1.on_third_id is None
|
|
|
|
play_1.hit = 1
|
|
|
|
assert play_1.hit == 1
|
|
|
|
play_1.pa, play_1.ab, play_1.hit, play_1.double, play_1.batter_final = 1, 1, 1, 1, 2
|
|
print(f'play_1: {play_1}')
|
|
|
|
play_2 = complete_play(session, play_1)
|
|
|
|
assert play_2.on_second is not None
|
|
|
|
|
|
def test_log_run_scored(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
lineup_1 = session.get(Lineup, 1)
|
|
play_1 = session.get(Play, 1)
|
|
|
|
assert play_1.run == 0
|
|
|
|
log_run_scored(session, lineup_1, play_1)
|
|
|
|
assert play_1.run == 1
|
|
|
|
play_2 = session.get(Play, 2)
|
|
play_2.pa, play_2.ab, play_2.double, play_2.batter_final = 1, 1, 1, 2
|
|
play_3 = complete_play(session, play_2)
|
|
|
|
assert play_3.on_second is not None
|
|
assert play_3.on_second.game == game_1
|
|
|
|
play_3.pa, play_3.ab, play_3.so, play_3.outs = 1, 1, 1, 1
|
|
play_3 = advance_runners(session, play_3, num_bases=0)
|
|
|
|
assert play_3.on_second_final == 2
|
|
|
|
play_4 = complete_play(session, play_3)
|
|
|
|
assert play_4.starting_outs == 2
|
|
assert play_4.on_second is not None
|
|
|
|
play_4.pa, play_4.ab, play_4.error = 1, 1, 1
|
|
log_run_scored(session, play_4.on_second, play_4)
|
|
|
|
runs = session.exec(select(func.sum(Play.run)).where(Play.game == game_1)).one()
|
|
e_runs = session.exec(select(func.sum(Play.e_run)).where(Play.game == game_1)).one()
|
|
|
|
assert runs == 2
|
|
assert e_runs == 1
|
|
|
|
|
|
async def test_strikeouts(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
play_1 = session.get(Play, 1)
|
|
|
|
play_1 = await strikeouts(session, None, play_1)
|
|
|
|
assert play_1.so == 1
|
|
assert play_1.outs == 1
|
|
assert play_1.batter_final is None
|
|
|
|
|
|
async def test_doubles(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
play_1 = session.get(Play, 1)
|
|
play_1.hit, play_1.batter_final = 1, 1
|
|
play_2 = complete_play(session, play_1)
|
|
|
|
assert play_2.play_num == 2
|
|
|
|
play_2_ghost_1 = await doubles(session, None, play_2, double_type='***')
|
|
|
|
assert play_2_ghost_1.double == 1
|
|
assert play_2_ghost_1.on_first_final == 4
|
|
assert play_2_ghost_1.rbi == 1
|
|
|
|
play_2_ghost_2 = await doubles(session, None, play_2, double_type='**')
|
|
|
|
assert play_2_ghost_2.double == 1
|
|
assert play_2_ghost_2.rbi == 0
|
|
assert play_2_ghost_2.on_first_final == 3
|
|
|
|
|
|
async def test_stealing(session: Session):
|
|
game_1 = session.get(Game, 1)
|
|
play_1 = session.get(Play, 1)
|
|
play_1.hit, play_1.batter_final = 1, 1
|
|
play_2 = complete_play(session, play_1)
|
|
|
|
assert play_2.play_num == 2
|
|
|
|
play_2 = await steals(session, None, play_2, 'stolen-base', to_base=2)
|
|
|
|
assert play_2.on_first_final == 2
|
|
assert play_2.sb == 1
|
|
assert play_2.runner == play_2.on_first
|
|
|
|
|
|
async def test_walks(session: Session):
|
|
"""Test walk functionality - both intentional and unintentional walks."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
# Test unintentional walk
|
|
play_1 = await walks(session, None, play_1, 'unintentional')
|
|
|
|
assert play_1.ab == 0 # No at-bat for walk
|
|
assert play_1.bb == 1 # Base on balls
|
|
assert play_1.ibb == 0 # Not intentional
|
|
assert play_1.batter_final == 1 # Batter reaches first
|
|
|
|
# Set up for intentional walk test
|
|
play_2 = complete_play(session, play_1)
|
|
play_2 = await walks(session, None, play_2, 'intentional')
|
|
|
|
assert play_2.ab == 0
|
|
assert play_2.bb == 1
|
|
assert play_2.ibb == 1 # Intentional walk
|
|
assert play_2.batter_final == 1
|
|
|
|
|
|
async def test_strikeouts(session: Session):
|
|
"""Test strikeout functionality."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
play_1 = await strikeouts(session, None, play_1)
|
|
|
|
assert play_1.so == 1 # Strikeout recorded
|
|
assert play_1.outs == 1 # One out recorded
|
|
# Note: batter_final is not set by strikeouts function, handled by advance_runners
|
|
|
|
|
|
async def test_popouts(session: Session):
|
|
"""Test popout functionality."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
play_1 = await popouts(session, None, play_1)
|
|
|
|
assert play_1.outs == 1 # One out recorded
|
|
# Note: batter_final is not set by popouts function, handled by advance_runners
|
|
|
|
|
|
async def test_hit_by_pitch(session: Session):
|
|
"""Test hit by pitch functionality."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
play_1 = await hit_by_pitch(session, None, play_1)
|
|
|
|
assert play_1.ab == 0 # No at-bat for HBP
|
|
assert play_1.hbp == 1 # Hit by pitch recorded
|
|
assert play_1.batter_final == 1 # Batter reaches first
|
|
|
|
|
|
async def test_homeruns(session: Session):
|
|
"""Test homerun functionality - both ballpark and no-doubt homers."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
# Test ballpark homerun
|
|
play_1 = await homeruns(session, None, play_1, 'ballpark')
|
|
|
|
assert play_1.hit == 1
|
|
assert play_1.homerun == 1
|
|
assert play_1.batter_final == 4 # Batter scores
|
|
assert play_1.run == 1 # Batter scores a run
|
|
assert play_1.rbi >= 1 # At least 1 RBI (for the batter themselves)
|
|
assert play_1.bphr == 1 # Ballpark homerun
|
|
|
|
# Test no-doubt homerun
|
|
play_2 = complete_play(session, play_1)
|
|
play_2 = await homeruns(session, None, play_2, 'no-doubt')
|
|
|
|
assert play_2.hit == 1
|
|
assert play_2.homerun == 1
|
|
assert play_2.batter_final == 4
|
|
assert play_2.run == 1
|
|
assert play_2.rbi >= 1
|
|
assert play_2.bphr == 0 # Not a ballpark homerun
|
|
|
|
|
|
async def test_singles(session: Session):
|
|
"""Test single functionality with different types."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
# Test regular single (*)
|
|
play_1 = await singles(session, None, play_1, '*')
|
|
|
|
assert play_1.hit == 1
|
|
assert play_1.batter_final == 1 # Batter reaches first
|
|
assert play_1.bp1b == 0 # Not a ballpark single
|
|
|
|
# Test ballpark single
|
|
play_2 = complete_play(session, play_1)
|
|
play_2 = await singles(session, None, play_2, 'ballpark')
|
|
|
|
assert play_2.hit == 1
|
|
assert play_2.batter_final == 1
|
|
assert play_2.bp1b == 1 # Ballpark single
|
|
|
|
# Test double-advance single (**)
|
|
play_3 = complete_play(session, play_2)
|
|
play_3 = await singles(session, None, play_3, '**')
|
|
|
|
assert play_3.hit == 1
|
|
assert play_3.batter_final == 1
|
|
|
|
|
|
async def test_doubles(session: Session):
|
|
"""Test double functionality with different types."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
# Test regular double (**)
|
|
play_1 = await doubles(session, None, play_1, '**')
|
|
|
|
assert play_1.hit == 1
|
|
assert play_1.double == 1
|
|
assert play_1.batter_final == 2 # Batter reaches second
|
|
|
|
# Test triple-advance double (***)
|
|
play_2 = complete_play(session, play_1)
|
|
play_2 = await doubles(session, None, play_2, '***')
|
|
|
|
assert play_2.hit == 1
|
|
assert play_2.double == 1
|
|
assert play_2.batter_final == 2
|
|
|
|
|
|
async def test_triples(session: Session):
|
|
"""Test triple functionality."""
|
|
this_game = session.get(Game, 3)
|
|
play_1 = this_game.initialize_play(session)
|
|
|
|
play_1 = await triples(session, None, play_1)
|
|
|
|
assert play_1.hit == 1
|
|
assert play_1.triple == 1
|
|
assert play_1.batter_final == 3 # Batter reaches third
|
|
|
|
|
|
async def test_bunts(session: Session):
|
|
"""Test bunt functionality with different types."""
|
|
this_game = session.get(Game, 3)
|
|
|
|
# Test sacrifice bunt
|
|
play_1 = this_game.initialize_play(session)
|
|
play_1 = await bunts(session, None, play_1, 'sacrifice')
|
|
|
|
assert play_1.ab == 0 # No at-bat for sacrifice
|
|
assert play_1.sac == 1 # Sacrifice recorded
|
|
assert play_1.outs == 1 # Out recorded
|
|
|
|
# Test bad bunt (batter reaches first safely)
|
|
play_2 = complete_play(session, play_1)
|
|
play_2 = await bunts(session, None, play_2, 'bad')
|
|
|
|
assert play_2.ab == 1 # At-bat recorded
|
|
assert play_2.sac == 0 # No sacrifice
|
|
assert play_2.outs == 1 # Out recorded
|
|
assert play_2.batter_final == 1 # Batter reaches first
|
|
|
|
# Test popout bunt
|
|
play_3 = complete_play(session, play_2)
|
|
play_3 = await bunts(session, None, play_3, 'popout')
|
|
|
|
assert play_3.ab == 1 # At-bat recorded
|
|
assert play_3.sac == 0 # No sacrifice
|
|
assert play_3.outs == 1 # Out recorded
|
|
|
|
# Test double-play bunt
|
|
play_4 = complete_play(session, play_3)
|
|
play_4 = await bunts(session, None, play_4, 'double-play')
|
|
|
|
assert play_4.ab == 1 # At-bat recorded
|
|
assert play_4.sac == 0 # No sacrifice
|
|
# Double play outs depend on starting_outs
|
|
|
|
|
|
async def test_chaos(session: Session):
|
|
"""Test chaos events - wild pitch, passed ball, balk, pickoff."""
|
|
this_game = session.get(Game, 3)
|
|
|
|
# Test wild pitch
|
|
play_1 = this_game.initialize_play(session)
|
|
play_1 = await chaos(session, None, play_1, 'wild-pitch')
|
|
|
|
assert play_1.pa == 0 # No plate appearance
|
|
assert play_1.ab == 0 # No at-bat
|
|
assert play_1.wild_pitch == 1 # Wild pitch recorded
|
|
assert play_1.rbi == 0 # No RBI on wild pitch
|
|
|
|
# Test passed ball
|
|
play_2 = complete_play(session, play_1)
|
|
play_2 = await chaos(session, None, play_2, 'passed-ball')
|
|
|
|
assert play_2.pa == 0
|
|
assert play_2.ab == 0
|
|
assert play_2.passed_ball == 1 # Passed ball recorded
|
|
assert play_2.rbi == 0
|
|
|
|
# Test balk
|
|
play_3 = complete_play(session, play_2)
|
|
play_3 = await chaos(session, None, play_3, 'balk')
|
|
|
|
assert play_3.pa == 0
|
|
assert play_3.ab == 0
|
|
assert play_3.balk == 1 # Balk recorded
|
|
assert play_3.rbi == 0
|
|
|
|
# Test pickoff
|
|
play_4 = complete_play(session, play_3)
|
|
play_4 = await chaos(session, None, play_4, 'pickoff')
|
|
|
|
assert play_4.pa == 0
|
|
assert play_4.ab == 0
|
|
assert play_4.pick_off == 1 # Pickoff recorded
|
|
assert play_4.outs == 1 # Out recorded
|