Add pitcher leaderboard

app.py

@@ -3,6 +3,7 @@ import matplotlib as mpl
 
 from data import data_df
 from pitcher_overview import create_pitcher_overview
+from pitcher_leaderboard import create_pitcher_leaderboard
 from pitch_leaderboard import create_pitch_leaderboard
 from daily_weekly_leaderboard import create_daily_weekly_leaderboard_app
 from css import css
@@ -15,6 +16,7 @@ with open('updated.txt') as f:
 
 limitations = '''**General Limitations**
 - As new players make their debut, some names may be missing or not translated/transliterated correctly.
+- IP is overestimated
 '''
 
 with open('acknowledgements.md', 'r') as f:
@@ -24,6 +26,8 @@ if __name__ == '__main__':
   with gr.Blocks(css=css) as app:
     with gr.Tab('Pitcher Overview'):
       create_pitcher_overview(data_df)
+    with gr.Tab('Pitcher Leaderboard'):
+      create_pitcher_leaderboard()
     with gr.Tab('Pitch Leaderboard'):
       create_pitch_leaderboard()
     with gr.Tab('Daily/Weekly Leaderboard'):

convert.py

@@ -161,6 +161,11 @@ presult = {
     141: 'Unknown'
 }
 
+def verify_and_return_presult(presults):
+  for _presult in presults:
+    assert _presult in presult.values(), f'{_presult} is invalid'
+  return presults
+
 bresult = {
     0: '空振り三振',
     1: '単打', # 1b gb p

pitch_leaderboard.py

@@ -32,6 +32,9 @@ TEAMS  = [
 ]
 notes = '''**Limitations**
 - [Column widths get messed up when filtering](https://github.com/gradio-app/gradio/issues/11564)
+
+**To-do**
+- Add post-season
 '''
 
 
@@ -44,29 +47,8 @@ def gr_create_pitch_leaderboard(start_date, end_date, min_pitches, pitcher_lr='B
   if pitcher_lr != 'Both':
     data = data.filter(pl.col('batLR') == pitcher_lr[0].lower())
     
-  # both, left, right = [
-    # (
-      # compute_pitch_stats(df, player_type='pitcher', min_pitches=min_pitches, pitch_class_type='specific')
-      # .filter(pl.col('qualified') & (pl.col('ballKind').is_in(include_pitches)))
-      # .drop('qualified')
-      # .rename({'pitcher_name': 'Pitcher', 'count': 'Count', 'usage': 'Usage', 'ballKind': 'Pitch', 'general_ballKind': 'Pitch (General)'} | {f'{stat}_pctl': f'{stat} (Pctl)' for stat in STATS_WITH_PCTLS})
-      # .with_columns(
-        # pl.col(stat).mul(100).round(1)
-        # for stat in PCT_STATS + [f'{stat} (Pctl)' for stat in STATS_WITH_PCTLS]
-      # )
-      # [['pitId', 'ballKind_code', 'Pitcher', 'Pitch', 'Pitch (General)', 'Count', 'Usage'] + STATS_WITH_PCTLS]
-    # )
-    # for df
-    # in [data, data.filter(pl.col('batLR') == 'l'), data.filter(pl.col('batLR') == 'r')]
-  # ]
-  # pitch_stats = (
-    # both
-    # .join(left, on=['pitId', 'ballKind_code'], suffix=' (LHH)', how='full')
-    # .join(right, on=['pitId', 'ballKind_code'], suffix=' (RHH)', how='full')
-    # .drop('pitId', 'ballKind_code', *list(chain.from_iterable([[f'{col} ({handedness}HH)' for col in ['pitId', 'ballKind_code', 'Pitcher', 'Pitch', 'Pitch (General)']] for handedness in ('L', 'R')])))
-  # )
   pitch_stats = (
-        compute_pitch_stats(data, player_type='pitcher', min_pitches=min_pitches, pitch_class_type='specific')
+        compute_pitch_stats(data, player_type='pitcher', min_pitches=min_pitches, pitch_class_type='specific', group_by_team=True)
         .filter(pl.col('qualified') & (pl.col('ballKind').is_in(include_pitches)))
         .drop('pitId', 'ballKind_code', 'qualified')
         .rename({
@@ -85,7 +67,6 @@ def gr_create_pitch_leaderboard(start_date, end_date, min_pitches, pitcher_lr='B
           pl.col(stat).mul(100)
           for stat in PCT_STATS
         )
-        # [['Pitcher', 'Team', 'Pitch', 'Pitch (General)'] + STATS + [f'{stat}_pctl' for stat in STATS_WITH_PCTLS]]
   )
 
   if include_teams is not None:
@@ -146,7 +127,7 @@ def create_pitch_leaderboard():
       with gr.Column(scale=1):
         all_pitches = gr.Button('Select/Deselect all pitches')
         min_pitches = gr.Number(100, label='Min. Pitches', precision=0, minimum=0)
-        pitcher_lr = gr.Radio(['Both', 'Left', 'Right'], value='Both', label='Batter handedness')
+        batter_lr = gr.Radio(['Both', 'Left', 'Right'], value='Both', label='Batter handedness')
     with gr.Row():
       include_teams = gr.CheckboxGroup(TEAMS, value=TEAMS, label='Teams', scale=3)
       all_teams = gr.Button('Select/Deselect all teams')
@@ -164,7 +145,7 @@ def create_pitch_leaderboard():
 
     gr.Markdown(notes)
 
-    search.click(gr_create_pitch_leaderboard, inputs=[start_date, end_date, min_pitches, pitcher_lr, include_pitches, include_teams], outputs=leaderboard)
+    search.click(gr_create_pitch_leaderboard, inputs=[start_date, end_date, min_pitches, batter_lr, include_pitches, include_teams], outputs=leaderboard)
     all_pitches.click(lambda _pitch_types : [] if _pitch_types == PITCH_TYPES else PITCH_TYPES, inputs=include_pitches, outputs=include_pitches)
     all_teams.click(lambda _teams : [] if _teams == TEAMS else TEAMS, inputs=include_teams, outputs=include_teams)
     # pin_columns.input(

pitcher_leaderboard.py

@@ -0,0 +1,158 @@
+import gradio as gr
+import polars as pl
+import numpy as np
+
+from datetime import datetime
+# from itertools import chain
+
+from data import data_df
+from stats import compute_player_stats, filter_data_by_date_and_game_kind
+from convert import team_names_short_to_color, get_text_color_from_team
+from plotting import stat_cmap
+
+STATS = ['FB Velo', 'K%', 'BB%', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Contact%', 'O-Contact%', 'SwStr%', 'Whiff%', 'CSW%', 'GB%', 'FB%', 'LD%', 'Zone%', 'Arm%', 'Glove%', 'High%', 'Low%', 'MM%']
+PCT_STATS = ['K%', 'BB%', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Contact%', 'O-Contact%', 'SwStr%', 'Whiff%', 'CSW%', 'GB%', 'FB%', 'LD%', 'Zone%', 'Arm%', 'Glove%', 'High%', 'Low%', 'MM%']
+STATS_WITH_PCTLS = ['FB Velo', 'K%', 'BB%', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Contact%', 'O-Contact%', 'SwStr%', 'Whiff%', 'CSW%', 'GB%', 'FB%', 'LD%', 'Zone%']
+COLUMNS = ['Pitcher', 'Team', 'IP', 'TBF'] + STATS
+
+TEAMS  = [
+    'Yomiuri',
+    'Yakult',
+    'DeNA',
+    'Chunichi',
+    'Hanshin',
+    'Hiroshima',
+    'Nipponham',
+    'Rakuten',
+    'Seibu',
+    'Lotte',
+    'ORIX',
+    'SoftBank'
+]
+notes = '''**Limitations**
+- [Column widths get messed up when filtering](https://github.com/gradio-app/gradio/issues/11564)
+
+**To-do**
+- Fix IP calculation
+- Add post-season
+'''
+
+
+def gr_create_pitcher_leaderboard(start_date, end_date, min_ip, pitcher_lr='Both', include_teams=None):
+  assert pitcher_lr in ['Both', 'Left', 'Right']
+
+  data = data_df.filter(pl.col('ballKind_code') != '-')
+  
+  data = filter_data_by_date_and_game_kind(data, start_date=start_date, end_date=end_date, game_kind='Regular Season')
+  if pitcher_lr != 'Both':
+    data = data.filter(pl.col('batLR') == pitcher_lr[0].lower())
+    
+  pitcher_stats = (
+        compute_player_stats(data, player_type='pitcher', min_ip=min_ip, group_by_team=True)
+        .filter(pl.col('qualified'))
+        .drop('pitId', 'qualified')
+        .rename({
+          'pitcher_name': 'Pitcher', 
+          'pitcher_team_name_short': 'Team', 
+          'PA': 'TBF'
+        })
+        .with_columns(
+          pl.col(stat).mul(100)
+          for stat in PCT_STATS
+        )
+  )
+
+  if include_teams is not None:
+    pitcher_stats = pitcher_stats.filter(pl.col('Team').is_in(include_teams))
+
+  styling = []
+  for i, row in enumerate(pitcher_stats[COLUMNS].iter_rows()):
+    styling_row = []
+    for col, item in zip(pitcher_stats[COLUMNS].columns, row):
+      _styling = 'font-size: 0.75em; '
+      if col in STATS_WITH_PCTLS:
+        r, g, b = (stat_cmap([pitcher_stats[f'{col}_pctl'][i]])[0, :3]*255).astype(np.uint8)
+        styling_row.append(f'background-color: rgba({r}, {g}, {b})')
+      elif col == 'Team':
+        styling_row.append(f'color: {get_text_color_from_team(item)}; background-color: {team_names_short_to_color[item]}')
+      else:
+        styling_row.append('')
+    styling.append(styling_row)
+
+  display_value = []
+  for row in pitcher_stats[COLUMNS].iter_rows():
+    display_value_row = []
+    for col, item in zip(COLUMNS, row):
+      if col in PCT_STATS:
+        display_value_row.append(f'{item:.1f}%')
+      elif isinstance(item, float):
+        display_value_row.append(f'{item:.1f}')
+      else:
+        display_value_row.append(item)
+    display_value.append(display_value_row)
+
+  value = {
+      'data': pitcher_stats[COLUMNS].rows(),
+      'headers': COLUMNS,
+      'metadata': {
+          'styling': styling,
+          'display_value': display_value,
+      }
+  }
+  
+  return value
+
+
+def create_pitcher_leaderboard():
+  now = datetime.now()
+  start_datetime_init = datetime(now.year, 1, 1)
+  end_datetime_init = now
+  with gr.Blocks() as app:
+    gr.Markdown('# Pitch Leaderboard')
+    with gr.Row():
+      start_date = gr.DateTime(start_datetime_init, include_time=False, type='datetime', label='Start')
+      end_date = gr.DateTime(end_datetime_init, include_time=False, type='datetime', label='End')
+    with gr.Row():
+      with gr.Group():
+        min_ip_state = gr.State('qualified')
+        min_ip = gr.Number(100, label='Min. IP', precision=0, minimum=0, interactive=False)
+        qualified = gr.Checkbox(True, label='Qualified')
+      batter_lr = gr.Radio(['Both', 'Left', 'Right'], value='Both', label='Batter handedness')
+    with gr.Row():
+      include_teams = gr.CheckboxGroup(TEAMS, value=TEAMS, label='Teams', scale=3)
+      all_teams = gr.Button('Select/Deselect all teams')
+        
+    search = gr.Button('Search')
+    pin_columns = gr.Button('Pin columns')
+    leaderboard = gr.DataFrame(
+      pl.DataFrame({'Pitcher': [], 'Pitch': []}),
+      column_widths=[125, 75, 50, 50] + [max(50, 10*len(stat)) for stat in STATS],
+      show_copy_button=True,
+      show_search='filter',
+      pinned_columns=2,
+      elem_id='leaderboard'
+    )
+
+    gr.Markdown(notes)
+
+    search.click(gr_create_pitcher_leaderboard, inputs=[start_date, end_date, min_ip_state, batter_lr, include_teams], outputs=leaderboard)
+    all_teams.click(lambda _teams : [] if _teams == TEAMS else TEAMS, inputs=include_teams, outputs=include_teams)
+    min_ip_state_kwargs = dict(fn=lambda min_ip, qualified: (qualified if qualified else min_ip, gr.Number(interactive=not qualified)), inputs=[min_ip, qualified], outputs=[min_ip_state, min_ip])
+    min_ip.change(**min_ip_state_kwargs)
+    qualified.change(**min_ip_state_kwargs)
+    # pin_columns.input(
+      # lambda _pin_columns : gr.update(pinned_columns=None if _pin_columns else 3),
+      # inputs=pin_columns,
+      # outputs=leaderboard
+    # )
+    # pin_columns_state = gr.State(True)
+    pin_columns.click(
+      lambda : gr.update(pinned_columns=None),
+      outputs=leaderboard
+    )
+    
+  return app
+
+if __name__ == '__main__':
+  app = create_pitcher_leaderboard()
+  app.launch()

plotting.py

@@ -92,7 +92,7 @@ def plot_loc(ax, locs):
 
 def plot_velo(ax, velos):
   trans = transforms.blended_transform_factory(ax.transData, ax.transAxes)
-  for (pitch,), _velos in velos.group_by('general_ballKind_code'):
+  for (pitch,), _velos in velos.sort(pl.len().over('general_ballKind_code'), descending=True).group_by('general_ballKind_code', maintain_order=True):
     _velos = _velos.filter(((pl.col('ballSpeed_mph') - pl.col('ballSpeed_mph').mean())/ pl.col('ballSpeed_mph').std()).abs() < 3)
 
     if len(_velos) <= 1:

stats.py

@@ -3,6 +3,12 @@ from data import data_df
 
 from types import SimpleNamespace
 
+from convert import verify_and_return_presult
+
+
+valid_pitch = pl.col('x').is_not_null() & pl.col('y').is_not_null() & (pl.col('ballSpeed') > 0)
+
+
 def filter_data_by_date_and_game_kind(data, start_date=None, end_date=None, game_kind=None):
   if start_date is not None:
     data = data.filter(pl.col('date') >= start_date)
@@ -63,17 +69,19 @@ def compute_team_games(data):
   )
 
 
-def compute_pitch_stats(data, player_type, pitch_class_type, min_pitches=1):
+def compute_pitch_stats(data, player_type, pitch_class_type, min_pitches=1, group_by_team=False):
   assert player_type in ('pitcher', 'batter')
   assert pitch_class_type in ('general', 'specific')
-  id_col = 'pitId' if player_type == 'pitcher' else 'batId'
+  id_cols = ['pitId' if player_type == 'pitcher' else 'batId']
+  if group_by_team:
+    id_cols.append('pitcher_team_name_short')
   name_col = 'pitcher_name' if player_type == 'pitcher' else 'batter_name'
   pitch_col = 'ballKind_code' if pitch_class_type == 'specific' else 'general_ballKind_code'
   pitch_name_col = 'ballKind' if pitch_class_type == 'specific' else 'general_ballKind'
   pitch_stats = (
       data
       .with_columns((pl.col('ballSpeed') / 1.609).round(1).alias('mph'))
-      .group_by(id_col, pitch_col, 'pitcher_team_name_short')
+      .group_by(*id_cols, pitch_col)
       .agg(
           pl.first(name_col),
           *([pl.first('general_ballKind')] if pitch_class_type == 'specific' else []),
@@ -119,17 +127,23 @@ def compute_pitch_stats(data, player_type, pitch_class_type, min_pitches=1):
           for stat in ['Avg KPH', 'Max KPH', 'Avg MPH', 'Max MPH', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Contact%', 'O-Contact%', 'SwStr%', 'Whiff%', 'CSW%', 'GB%', 'FB%', 'LD%', 'Zone%']
       )
       .rename({pitch_col: 'ballKind_code', pitch_name_col: 'ballKind'} if pitch_class_type == 'general' else {})
-      .sort(id_col, 'count', descending=[False, True])
+      .sort(id_cols[0], 'count', descending=[False, True])
   )
   return pitch_stats
-  
 
-def compute_player_stats(data, player_type, min_ip='qualified'):
+def compute_player_stats(data, player_type, min_ip='qualified', group_by_team=False):
   data = (
       compute_team_games(data)
       .with_columns(
           pl.when(pl.col('half_inning').str.ends_with('1')).then('home_games').otherwise('visitor_games').first().over('pitId').alias('games'),
           pl.col('inning_code').unique().len().over('pitId').alias('IP')
+          # pl.col('presult').is_in(verify_and_return_presult([
+            # 'Groundout', 'Flyout', 'Lineout', 'Groundout (Double play)',
+            # 'Foul fly', 'Foul line (?)',
+            # 'Sacrifice bunt', 'Sacrifice fly',
+            # "Fielder's choice", "Sacrifice fielder's choice",
+            # 'Bunt strikeout', 'Swinging strikeout', 'Looking strikeout'
+          # ])).sum().over('pitId').mul(1/3).alias('IP')
       )
   )
 
@@ -139,17 +153,38 @@ def compute_player_stats(data, player_type, min_ip='qualified'):
     data = data.with_columns((pl.col('IP') >= min_ip).alias('qualified'))
 
   assert player_type in ('pitcher', 'batter') 
-  id_col = 'pitId' if player_type == 'pitcher' else 'batId'
+  id_cols = ['pitId' if player_type == 'pitcher' else 'batId']
+  if group_by_team:
+    id_cols.append('pitcher_team_name_short')
   name_col = 'pitcher_name' if player_type == 'pitcher' else 'batter_name'
   player_stats = (
     data
-    .group_by(id_col)
+    .with_columns(pl.when(pl.col('general_ballKind_code').is_in(['4S', 'FC', 'SI'])).then(pl.when(valid_pitch).then('ballSpeed').mean().over('pitId', 'general_ballKind_code')).mul(1/1.609).round(1).alias('FB Velo'))
+    .group_by(id_cols)
     .agg(
         pl.col(name_col).first(),
+        *([] if group_by_team else [pl.col('pitcher_team_name_short').last()]),
+        pl.col('IP').first(),
+        pl.col('pa_code').unique().len().alias('PA'),
+        pl.col('FB Velo').max(),
         (pl.when(pl.col('presult').str.contains('strikeout')).then(1).otherwise(0).sum() / pl.col('pa_code').unique().len()).alias('K%'),
         (pl.when(pl.col('presult') == 'Walk').then(1).otherwise(0).sum() / pl.col('pa_code').unique().len()).alias('BB%'),
-        (pl.col('csw').sum() / pl.col('pitch').sum()).alias('CSW%'),
         pl.col('aux_bresult').struct.field('batType').drop_nulls().value_counts(normalize=True),
+        (pl.col('swing').sum() / pl.col('pitch').sum()).alias('Swing%'),
+        ((pl.col('swing') & pl.col('zone')).sum() / pl.col('pitch').sum()).alias('Z-Swing%'),
+        ((pl.col('swing') & ~pl.col('zone')).sum() / pl.col('pitch').sum()).alias('Chase%'),
+        ((pl.col('swing') & ~pl.col('whiff')).sum()/pl.col('swing').sum()).alias('Contact%'),
+        ((pl.col('zone') & pl.col('swing') & ~pl.col('whiff')).sum()/(pl.col('zone') & pl.col('swing')).sum()).alias('Z-Contact%'),
+        ((~pl.col('zone') & pl.col('swing') & ~pl.col('whiff')).sum()/(~pl.col('zone') & pl.col('swing')).sum()).alias('O-Contact%'),
+        (pl.col('whiff').sum() / pl.col('swing').sum()).alias('Whiff%'),
+        (pl.col('whiff').sum() / pl.col('pitch').sum()).alias('SwStr%'),
+        (pl.col('csw').sum() / pl.col('pitch').sum()).alias('CSW%'),
+        (pl.col('zone').sum() / pl.col('pitch').sum()).alias('Zone%'),
+        (pl.when(pl.col('pitLR') == 'r').then(pl.col('x') < 0).otherwise(pl.col('x') > 0)).mean().alias('Glove%'),
+        (pl.when(pl.col('pitLR') == 'r').then(pl.col('x') >= 0).otherwise(pl.col('x') <= 0)).mean().alias('Arm%'),
+        (pl.col('y') > 125).mean().alias('High%'),
+        (pl.col('y') <= 125).mean().alias('Low%'),
+        (pl.col('x').is_between(-20, 20) & pl.col('y').is_between(100, 100+50)).mean().alias('MM%'),
         pl.first('qualified')
     )
     .explode('batType')
@@ -163,9 +198,10 @@ def compute_player_stats(data, player_type, min_ip='qualified'):
     )
     .drop('G', 'F', 'B', 'P', 'L')
     .with_columns(
-        (pl.when(pl.col('qualified')).then(pl.col(stat)).rank(descending=(stat == 'BB%'))/pl.when(pl.col('qualified')).then(pl.col(stat)).count()).alias(f'{stat}_pctl')
-        for stat in ['CSW%', 'K%', 'BB%', 'GB%']
+        (pl.when(pl.col('qualified')).then(pl.col(stat)).rank(descending=stat in ['BB%', 'FB%', 'LD%'] or 'Contact%' in stat)/pl.when(pl.col('qualified')).then(pl.col(stat)).count()).alias(f'{stat}_pctl')
+        for stat in ['FB Velo', 'K%', 'BB%', 'Swing%', 'Z-Swing%', 'Chase%', 'Contact%', 'Z-Contact%', 'O-Contact%', 'SwStr%', 'Whiff%', 'CSW%', 'GB%', 'FB%', 'LD%', 'Zone%']
     )
+    .sort('IP', descending=True)
   )
   return player_stats
 
@@ -186,7 +222,7 @@ def get_pitcher_stats(id, lr=None, game_kind=None, start_date=None, end_date=Non
   if lr is not None:
     source_data = source_data.filter(pl.col('batLR') == lr)
 
-  pitch_stats = compute_pitch_stats(source_data, player_type='pitcher', pitch_class_type=pitch_class_type, min_pitches=min_pitches).filter(pl.col('pitId') == id)
+  pitch_stats = compute_pitch_stats(source_data, player_type='pitcher', pitch_class_type=pitch_class_type, min_pitches=min_pitches, group_by_team=False).filter(pl.col('pitId') == id)
 
   pitch_shapes = (
       source_data
@@ -200,6 +236,6 @@ def get_pitcher_stats(id, lr=None, game_kind=None, start_date=None, end_date=Non
       .with_columns((pl.col('ballSpeed')/1.609).alias('ballSpeed_mph'))
   )
 
-  pitcher_stats = compute_player_stats(source_data, player_type='pitcher', min_ip=min_ip).filter(pl.col('pitId') == id)
+  pitcher_stats = compute_player_stats(source_data, player_type='pitcher', min_ip=min_ip, group_by_team=False).filter(pl.col('pitId') == id)
 
   return SimpleNamespace(pitcher_stats=pitcher_stats, pitch_stats=pitch_stats, pitch_shapes=pitch_shapes)