Calculating wpa for plays

command_logic/logic_gameplay.py

@@ -1,6 +1,7 @@
 
 import logging
 import discord
+import pandas as pd
 from sqlmodel import Session, select
 from typing import Literal
 
@@ -13,13 +14,84 @@ from utilities.embeds import image_embed
 from utilities.pages import Pagination
 
 
+WPA_DF = pd.read_csv(f'storage/wpa_data.csv').set_index('index')
+
+
+def get_obc(on_first = None, on_second = None, on_third = None) -> int:
+    if on_third is not None:
+        if on_second is not None:
+            if on_first is not None:
+                obc = 7
+            else:
+                obc = 6
+        elif on_first is not None:
+            obc = 5
+        else:
+            obc = 3
+    elif on_second is not None:
+        if on_first is not None:
+            obc = 4
+        else:
+            obc = 2
+    elif on_first is not None:
+        obc = 1
+    else:
+        obc = 0
+
+    return obc        
+
+
+def get_re24(this_play: Play, runs_scored: int, new_obc: int, new_starting_outs: int) -> float:
+    re_data = {
+        0: [0.457, 0.231, 0.077],
+        1: [0.793, 0.438, 0.171],
+        2: [1.064, 0.596, 0.259],
+        4: [1.373, 0.772, 0.351],
+        3: [1.340, 0.874, 0.287],
+        5: [1.687, 1.042, 0.406],
+        6: [1.973, 1.311, 0.448],
+        7: [2.295, 1.440, 0.618]
+    }
+
+    start_re24 = re_data[this_play.on_base_code][this_play.starting_outs]
+    end_re24 = 0 if this_play.starting_outs + this_play.outs > 2 else re_data[new_obc][new_starting_outs]
+    return round(end_re24 - start_re24 + runs_scored, 3)
+
+
+def get_wpa(old_play: Play, new_play: Play):
+    new_rd = new_play.home_score - new_play.away_score
+    if new_rd > 6:
+        new_rd = 6
+    elif new_rd < -6:
+        new_rd = -6
+
+    old_rd = old_play.home_score - old_play.away_score
+    if old_rd > 6:
+        old_rd = 6
+    elif old_rd < -6:
+        old_rd = -6
+
+    new_win_ex = WPA_DF.loc[f'{new_play.inning_half}_{new_play.inning_num}_{new_play.starting_outs}_out_{new_play.on_base_code}_obc_{new_rd}_home_run_diff'].home_win_ex
+
+    old_win_ex = WPA_DF.loc[f'{old_play.inning_half}_{old_play.inning_num}_{old_play.starting_outs}_out_{old_play.on_base_code}_obc_{old_rd}_home_run_diff'].home_win_ex
+
+    return round(new_win_ex - old_win_ex, 3)
+
+
 def complete_play(session:Session, this_play: Play):
     nso = this_play.starting_outs + this_play.outs
-    
+    runs_scored = 0
+    on_first, on_second, on_third = None, None, None
+
     if nso >= 3:
         switch_sides = True
+        obc = 0
         nso = 0
+        is_go_ahead = False
         nih = 'bot' if this_play.inning_half.lower() == 'top' else 'top'
+        away_score = this_play.away_score
+        home_score = this_play.home_score
+
         try:
             opponent_play = get_last_team_play(session, this_play.game, this_play.pitcher.team)
             nbo = opponent_play.batting_order + 1
@@ -29,28 +101,78 @@ def complete_play(session:Session, this_play: Play):
         finally:
             new_batter_team = this_play.game.away_team if nih == 'top' else this_play.game.home_team
             new_pitcher_team = this_play.game.away_team if nih == 'bot' else this_play.game.home_team
+            inning = this_play.inning_num if nih == 'bot' else this_play.inning_num + 1
+            
     else:
         switch_sides = False
         nbo = this_play.batting_order + 1 if this_play.pa == 1 else this_play.batting_order
         nih = this_play.inning_half
         new_batter_team = this_play.batter.team
         new_pitcher_team = this_play.pitcher.team
+        inning = this_play.inning_num
+    
+        for this_runner, runner_dest in [
+            (this_play.batter, this_play.batter_final), (this_play.on_first, this_play.on_first_final), (this_play.on_second, this_play.on_second_final), (this_play.on_third, this_play.on_third_final)
+        ]:
+            if runner_dest is not None:
+                if runner_dest == 1:
+                    if on_first is not None:
+                        log_exception(ValueError, f'Cannot place {this_runner.player.name} on first; {on_first.player.name} is already placed there')
+                    on_first = this_runner
+                elif runner_dest == 2:
+                    if on_second is not None:
+                        log_exception(ValueError, f'Cannot place {this_runner.player.name} on second; {on_second.player.name} is already placed there')
+                    on_second = this_runner
+                elif runner_dest == 3:
+                    if on_third is not None:
+                        log_exception(ValueError, f'Cannot place {this_runner.player.name} on third; {on_third.player.name} is already placed there')
+                    on_third = this_runner
+                elif runner_dest == 4:
+                    runs_scored += 1
+        
+        if this_play.inning_half == 'top':
+            away_score = this_play.away_score + runs_scored
+            home_score = this_play.home_score
+            
+            if runs_scored > 0 and this_play.away_score <= this_play.home_score and away_score > home_score:
+                is_go_ahead = True
+            
+        else:
+            away_score = this_play.away_score
+            home_score = this_play.home_score + runs_scored
+
+            if runs_scored > 0 and this_play.home_score <= this_play.away_score and home_score > away_score:
+                is_go_ahead = True        
+        
+        obc = get_obc(on_first, on_second, on_third)
+    
+    this_play.re24 = get_re24(this_play, runs_scored, new_obc=obc, new_starting_outs=nso)
 
     if nbo > 9:
         nbo = 1
-    
+
-    # TODO: Set baserunners
+    new_batter = get_one_lineup(session, this_play.game, new_batter_team, batting_order=nbo)
 
     new_play = Play(
         game=this_play.game, 
         play_num=this_play.play_num + 1,
         batting_order=nbo,
         inning_half=nih,
+        inning_num=inning,
         starting_outs=nso,
-        batter=get_one_lineup(session, this_play.game, new_batter_team, batting_order=nbo),
+        on_base_code=obc,
+        away_score=away_score,
+        home_score=home_score,
+        batter=new_batter,
+        batter_pos=new_batter.position,
         pitcher=get_one_lineup(session, this_play.game, new_pitcher_team, position='P'),
         catcher=get_one_lineup(session, this_play.game, new_pitcher_team, position='C'),
         is_new_inning=switch_sides,
+        is_tied=away_score == home_score,
+        is_go_ahead=is_go_ahead,
+        on_first=on_first,
+        on_second=on_second,
+        on_third=on_third,
         managerai=this_play.managerai
     )
 
@@ -459,11 +581,10 @@ async def flyballs(session: Session, interaction: discord.Interaction, this_game
                 await question.delete()
 
     elif flyball_type == 'c':
-        patch_play(this_play.id, locked=True, pa=1, ab=1, outs=1)
+        this_play.pa, this_play.ab, this_play.outs = 1, 1, 1
-        advance_runners(this_play.id, num_bases=0)
 
     if comp_play:
-        complete_play(this_play.id)
+        complete_play(session, this_play)
     
     session.refresh(this_play)
     return this_play

in_game/gameplay_models.py

@@ -568,12 +568,12 @@ class PlayBase(SQLModel):
     in_pow: bool = Field(default=False)
 
     on_first_id: int | None = Field(default=None, foreign_key='lineup.id')
-    on_first_final: int | None = Field(default=None)
+    on_first_final: int | None = Field(default=None)                    # 99 = out, 1-4 = base, None = no change
     on_second_id: int | None = Field(default=None, foreign_key='lineup.id')
-    on_second_final: int | None = Field(default=None)
+    on_second_final: int | None = Field(default=None)                   # 99 = out, 1-4 = base, None = no change
     on_third_id: int | None = Field(default=None, foreign_key='lineup.id')
-    on_third_final: int | None = Field(default=None)
+    on_third_final: int | None = Field(default=None)                    # 99 = out, 1-4 = base, None = no change
-    batter_final: int | None = Field(default=None)
+    batter_final: int | None = Field(default=None)                      # 99 = out, 1-4 = base, None = out
 
     pa: int = Field(default=0, ge=0, le=1)
     ab: int = Field(default=0, ge=0, le=1)

requirements.txt

@@ -8,3 +8,4 @@ sqlmodel
 alembic
 pytest
 pytest-asyncio
+pandas

tests/command_logic/test_logic_gameplay.py

@@ -1,7 +1,7 @@
 import pytest
 from sqlmodel import Session
 
-from command_logic.logic_gameplay import advance_runners
+from command_logic.logic_gameplay import advance_runners, get_obc, get_re24, get_wpa
 from tests.factory import session_fixture, Game
 
 
@@ -15,3 +15,33 @@ def test_advance_runners(session: Session):
     assert play_1.on_third_id is None
 
     # TODO: Test advance runners once "advance play" function is ready
+
+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_re24(session: Session):
+    game_1 = session.get(Game, 1)
+    this_play = game_1.current_play_or_none(session)
+    old_play = game_1.plays[0]
+
+    assert old_play.id != this_play
+