from fastapi import APIRouter, Depends, HTTPException, Query from typing import List, Optional, Literal import copy import logging from pydantic import BaseModel, validator from ..db_engine import db, StratPlay, StratGame, Team, Player, Decision, model_to_dict, chunked, fn, SQL from ..dependencies import oauth2_scheme, valid_token, LOG_DATA logging.basicConfig( filename=LOG_DATA['filename'], format=LOG_DATA['format'], level=LOG_DATA['log_level'] ) router = APIRouter( prefix='/api/v3/plays', tags=['plays'] ) POS_LIST = Literal['C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF', 'P', 'DH', 'PH', 'PR'] class PlayModel(BaseModel): game_id: int play_num: int batter_id: int = None batter_team_id: int = None pitcher_id: int pitcher_team_id: int = None on_base_code: str inning_half: Literal['top', 'bot', 'Top', 'Bot'] inning_num: int batting_order: int starting_outs: int away_score: int home_score: int batter_pos: POS_LIST = None on_first_id: int = None on_first_final: int = None on_second_id: int = None on_second_final: int = None on_third_id: int = None on_third_final: int = None batter_final: int = None pa: int = 0 ab: int = 0 e_run: int = 0 run: int = 0 hit: int = 0 rbi: int = 0 double: int = 0 triple: int = 0 homerun: int = 0 bb: int = 0 so: int = 0 hbp: int = 0 sac: int = 0 ibb: int = 0 gidp: int = 0 bphr: int = 0 bpfo: int = 0 bp1b: int = 0 bplo: int = 0 sb: int = 0 cs: int = 0 outs: int = 0 wpa: float = 0 catcher_id: int = None catcher_team_id: int = None defender_id: int = None defender_team_id: int = None runner_id: int = None runner_team_id: int = None check_pos: POS_LIST = None error: int = 0 wild_pitch: int = 0 passed_ball: int = 0 pick_off: int = 0 balk: int = 0 is_go_ahead: bool = False is_tied: bool = False is_new_inning: bool = False @validator('on_first_final') def no_final_if_no_runner_one(cls, v, values): if values['on_first_id'] is None: return None return v @validator('on_second_final') def no_final_if_no_runner_two(cls, v, values): if values['on_second_id'] is None: return None return v @validator('on_third_final') def no_final_if_no_runner_three(cls, v, values): if values['on_third_id'] is None: return None return v @validator('batter_final') def no_final_if_no_batter(cls, v, values): if values['batter_id'] is None: return None return v class PlayList(BaseModel): plays: List[PlayModel] @router.get('') async def get_plays( game_id: list = Query(default=None), batter_id: list = Query(default=None), season: list = Query(default=None), week: list = Query(default=None), has_defender: Optional[bool] = None, has_catcher: Optional[bool] = None, has_defender_or_catcher: Optional[bool] = None, is_scoring_play: Optional[bool] = None, pitcher_id: list = Query(default=None), obc: list = Query(default=None), inning: list = Query(default=None), batting_order: list = Query(default=None), starting_outs: list = Query(default=None), batter_pos: list = Query(default=None), catcher_id: list = Query(default=None), defender_id: list = Query(default=None), runner_id: list = Query(default=None), offense_team_id: list = Query(default=None), defense_team_id: list = Query(default=None), hit: Optional[int] = None, double: Optional[int] = None, triple: Optional[int] = None, homerun: Optional[int] = None, sb: Optional[int] = None, cs: Optional[int] = None, manager_id: list = Query(default=None), run: Optional[int] = None, e_run: Optional[int] = None, rbi: list = Query(default=None), outs: list = Query(default=None), wild_pitch: Optional[int] = None, is_final_out: Optional[bool] = None, is_go_ahead: Optional[bool] = None, is_tied: Optional[bool] = None, is_new_inning: Optional[bool] = None, min_wpa: Optional[float] = None, max_wpa: Optional[float] = None, short_output: Optional[bool] = False, sort: Optional[str] = None, limit: Optional[int] = 200): all_plays = StratPlay.select() if season is not None: s_games = StratGame.select().where(StratGame.season << season) all_plays = all_plays.where(StratPlay.game << s_games) if week is not None: w_games = StratGame.select().where(StratGame.week << week) all_plays = all_plays.where(StratPlay.game << w_games) if has_defender is not None: all_plays = all_plays.where(StratPlay.defender.is_null(False)) if has_catcher is not None: all_plays = all_plays.where(StratPlay.catcher.is_null(False)) if has_defender_or_catcher is not None: all_plays = all_plays.where( (StratPlay.catcher.is_null(False)) | (StratPlay.defender.is_null(False)) ) if game_id is not None: all_plays = all_plays.where(StratPlay.game_id << game_id) if batter_id is not None: all_plays = all_plays.where(StratPlay.batter_id << batter_id) if pitcher_id is not None: all_plays = all_plays.where(StratPlay.pitcher_id << pitcher_id) if obc is not None: all_plays = all_plays.where(StratPlay.on_base_code << obc) if inning is not None: all_plays = all_plays.where(StratPlay.inning_num << inning) if batting_order is not None: all_plays = all_plays.where(StratPlay.batting_order << batting_order) if starting_outs is not None: all_plays = all_plays.where(StratPlay.starting_outs << starting_outs) if batter_pos is not None: all_plays = all_plays.where(StratPlay.batter_pos << batter_pos) if catcher_id is not None: all_plays = all_plays.where(StratPlay.catcher_id << catcher_id) if defender_id is not None: all_plays = all_plays.where(StratPlay.defender_id << defender_id) if runner_id is not None: all_plays = all_plays.where(StratPlay.runner_id << runner_id) if offense_team_id is not None: all_teams = Team.select().where(Team.id << offense_team_id) all_plays = all_plays.where( (StratPlay.batter_team << all_teams) | (StratPlay.runner_team << all_teams) ) if defense_team_id is not None: all_teams = Team.select().where(Team.id << defense_team_id) all_plays = all_plays.where( (StratPlay.catcher_team << all_teams) | (StratPlay.defender_team << all_teams) ) if hit is not None: all_plays = all_plays.where(StratPlay.hit == hit) if double is not None: all_plays = all_plays.where(StratPlay.double == double) if triple is not None: all_plays = all_plays.where(StratPlay.triple == triple) if homerun is not None: all_plays = all_plays.where(StratPlay.homerun == homerun) if sb is not None: all_plays = all_plays.where(StratPlay.sb == sb) if cs is not None: all_plays = all_plays.where(StratPlay.cs == cs) if wild_pitch is not None: all_plays = all_plays.where(StratPlay.wild_pitch == wild_pitch) if run is not None: all_plays = all_plays.where(StratPlay.run == run) if e_run is not None: all_plays = all_plays.where(StratPlay.e_run == e_run) if rbi is not None: all_plays = all_plays.where(StratPlay.rbi << rbi) if outs is not None: all_plays = all_plays.where(StratPlay.outs << outs) if manager_id is not None: all_games = StratGame.select().where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) all_plays = all_plays.where(StratPlay.game << all_games) if is_final_out is not None: all_plays = all_plays.where(StratPlay.starting_outs + StratPlay.outs == 3) if is_go_ahead is not None: all_plays = all_plays.where(StratPlay.is_go_ahead == is_go_ahead) if is_tied is not None: all_plays = all_plays.where(StratPlay.is_tied == is_tied) if is_new_inning is not None: all_plays = all_plays.where(StratPlay.is_new_inning == is_new_inning) if is_scoring_play is not None: all_plays = all_plays.where( (StratPlay.on_first_final == 4) | (StratPlay.on_second_final == 4) | (StratPlay.on_third_final == 4) | (StratPlay.batter_final == 4) ) if min_wpa is not None: all_plays = all_plays.where(StratPlay.wpa >= min_wpa) if max_wpa is not None: all_plays = all_plays.where(StratPlay.wpa <= max_wpa) if limit > 5000: limit = 5000 elif limit < 1: limit = 1 if sort == 'wpa-desc': all_plays = all_plays.order_by(-fn.ABS(StratPlay.wpa)) elif sort == 'wpa-asc': all_plays = all_plays.order_by(fn.ABS(StratPlay.wpa)) elif sort == 'newest': all_plays = all_plays.order_by(StratPlay.game_id.desc(), StratPlay.play_num.desc()) elif sort == 'oldest': all_plays = all_plays.order_by(StratPlay.game_id, StratPlay.play_num) all_plays = all_plays.limit(limit) return_plays = { 'count': all_plays.count(), 'plays': [model_to_dict(x, recurse=not short_output) for x in all_plays] } db.close() return return_plays @router.get('/batting') async def get_batting_totals( season: list = Query(default=None), week: list = Query(default=None), s_type: Literal['regular', 'post', 'total', None] = None, position: list = Query(default=None), player_id: list = Query(default=None), min_wpa: Optional[float] = -999, max_wpa: Optional[float] = 999, group_by: Literal['team', 'player', 'playerteam', 'playergame', 'teamgame', 'league'] = 'player', min_pa: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), obc: list = Query(default=None), risp: Optional[bool] = None, inning: list = Query(default=None), sort: Optional[str] = None, limit: Optional[int] = None, short_output: Optional[bool] = False): season_games = StratGame.select() if season is not None: season_games = season_games.where(StratGame.season << season) if week is not None and s_type is not None: raise HTTPException(status_code=400, detail=f'Week and s_type parameters cannot be used in the same query') if week is not None: season_games = season_games.where(StratGame.week << week) if s_type is not None: if s_type == 'regular': season_games = season_games.where(StratGame.week <= 18) elif s_type == 'post': season_games = season_games.where(StratGame.week > 18) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) bat_plays = ( StratPlay .select(StratPlay.batter, StratPlay.game, fn.SUM(StratPlay.pa).alias('sum_pa'), fn.SUM(StratPlay.ab).alias('sum_ab'), fn.SUM(StratPlay.run).alias('sum_run'), fn.SUM(StratPlay.hit).alias('sum_hit'), fn.SUM(StratPlay.rbi).alias('sum_rbi'), fn.SUM(StratPlay.double).alias('sum_double'), fn.SUM(StratPlay.triple).alias('sum_triple'), fn.SUM(StratPlay.homerun).alias('sum_hr'), fn.SUM(StratPlay.bb).alias('sum_bb'), fn.SUM(StratPlay.so).alias('sum_so'), StratPlay.batter_team, fn.SUM(StratPlay.hbp).alias('sum_hbp'), fn.SUM(StratPlay.sac).alias('sum_sac'), fn.SUM(StratPlay.ibb).alias('sum_ibb'), fn.SUM(StratPlay.gidp).alias('sum_gidp'), fn.SUM(StratPlay.sb).alias('sum_sb'), fn.SUM(StratPlay.cs).alias('sum_cs'), fn.SUM(StratPlay.bphr).alias('sum_bphr'), fn.SUM(StratPlay.bpfo).alias('sum_bpfo'), fn.SUM(StratPlay.bp1b).alias('sum_bp1b'), fn.SUM(StratPlay.bplo).alias('sum_bplo'), fn.SUM(StratPlay.wpa).alias('sum_wpa'), fn.COUNT(StratPlay.on_first_final).filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4)).alias('count_lo1'), fn.COUNT(StratPlay.on_second_final).filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4)).alias('count_lo2'), fn.COUNT(StratPlay.on_third_final).filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4)).alias('count_lo3'), fn.COUNT(StratPlay.on_first).filter(StratPlay.on_first.is_null(False)).alias('count_runner1'), fn.COUNT(StratPlay.on_second).filter(StratPlay.on_second.is_null(False)).alias('count_runner2'), fn.COUNT(StratPlay.on_third).filter(StratPlay.on_third.is_null(False)).alias('count_runner3'), fn.COUNT(StratPlay.on_first_final).filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_lo1_3out'), fn.COUNT(StratPlay.on_second_final).filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_lo2_3out'), fn.COUNT(StratPlay.on_third_final).filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_lo3_3out') # fn.COUNT(StratPlay.on_first).filter(StratPlay.on_first.is_null(False) & # (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_runner1_3out'), # fn.COUNT(StratPlay.on_second).filter(StratPlay.on_second.is_null(False) & # (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_runner2_3out'), # fn.COUNT(StratPlay.on_third).filter(StratPlay.on_third.is_null(False) & # (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_runner3_3out') ) .where((StratPlay.game << season_games) & (StratPlay.batter.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_pa) ) run_plays = ( StratPlay .select(StratPlay.runner, StratPlay.runner_team, fn.SUM(StratPlay.sb).alias('sum_sb'), fn.SUM(StratPlay.cs).alias('sum_cs'), fn.SUM(StratPlay.pick_off).alias('sum_pick'), fn.SUM(StratPlay.wpa).alias('sum_wpa')) .where((StratPlay.game << season_games) & (StratPlay.runner.is_null(False))) ) def_plays = ( StratPlay .select(StratPlay.defender, StratPlay.defender_team, fn.SUM(StratPlay.error).alias('sum_error'), fn.SUM(StratPlay.hit).alias('sum_hit'), fn.SUM(StratPlay.pa).alias('sum_chances'), fn.SUM(StratPlay.wpa).alias('sum_wpa')) .where((StratPlay.game << season_games) & (StratPlay.defender.is_null(False))) ) if player_id is not None: all_players = Player.select().where(Player.id << player_id) bat_plays = bat_plays.where(StratPlay.batter << all_players) run_plays = run_plays.where(StratPlay.runner << all_players) def_plays = def_plays.where(StratPlay.defender << all_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) bat_plays = bat_plays.where(StratPlay.batter_team << all_teams) run_plays = run_plays.where(StratPlay.runner_team << all_teams) def_plays = def_plays.where(StratPlay.defender_team << all_teams) if position is not None: bat_plays = bat_plays.where(StratPlay.batter_pos << position) if obc is not None: bat_plays = bat_plays.where(StratPlay.on_base_code << obc) if risp is not None: bat_plays = bat_plays.where(StratPlay.on_base_code << ['100', '101', '110', '111', '010', '011']) if inning is not None: bat_plays = bat_plays.where(StratPlay.inning_num << inning) if group_by is not None: if group_by == 'player': bat_plays = bat_plays.group_by(StratPlay.batter) run_plays = run_plays.group_by(StratPlay.runner) def_plays = def_plays.group_by(StratPlay.defender) elif group_by == 'team': bat_plays = bat_plays.group_by(StratPlay.batter_team) run_plays = run_plays.group_by(StratPlay.runner_team) def_plays = def_plays.group_by(StratPlay.defender_team) elif group_by == 'playerteam': bat_plays = bat_plays.group_by(StratPlay.batter, StratPlay.batter_team) run_plays = run_plays.group_by(StratPlay.runner, StratPlay.runner_team) def_plays = def_plays.group_by(StratPlay.defender, StratPlay.defender_team) elif group_by == 'playergame': bat_plays = bat_plays.group_by(StratPlay.batter, StratPlay.game) run_plays = run_plays.group_by(StratPlay.runner, StratPlay.game) def_plays = def_plays.group_by(StratPlay.defender, StratPlay.game) elif group_by == 'teamgame': bat_plays = bat_plays.group_by(StratPlay.batter_team, StratPlay.game) run_plays = run_plays.group_by(StratPlay.runner_team, StratPlay.game) def_plays = def_plays.group_by(StratPlay.defender_team, StratPlay.game) elif group_by == 'league': bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by(StratPlay.game.season) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by(StratPlay.game.season) if sort is not None: if sort == 'player': bat_plays = bat_plays.order_by(StratPlay.batter) run_plays = run_plays.order_by(StratPlay.runner) def_plays = def_plays.order_by(StratPlay.defender) elif sort == 'team': bat_plays = bat_plays.order_by(StratPlay.batter_team) run_plays = run_plays.order_by(StratPlay.runner_team) def_plays = def_plays.order_by(StratPlay.defender_team) elif sort == 'wpa-desc': bat_plays = bat_plays.order_by(SQL('sum_wpa').desc()) elif sort == 'wpa-asc': bat_plays = bat_plays.order_by(SQL('sum_wpa').asc()) elif sort == 'pa-desc': bat_plays = bat_plays.order_by(SQL('sum_pa').desc()) elif sort == 'pa-asc': bat_plays = bat_plays.order_by(SQL('sum_pa').asc()) elif sort == 'newest': bat_plays = bat_plays.order_by(StratPlay.game_id.desc(), StratPlay.play_num.desc()) run_plays = run_plays.order_by(StratPlay.game_id.desc(), StratPlay.play_num.desc()) elif sort == 'oldest': bat_plays = bat_plays.order_by(StratPlay.game_id, StratPlay.play_num) run_plays = run_plays.order_by(StratPlay.game_id, StratPlay.play_num) if limit is not None: if limit < 1: limit = 1 bat_plays = bat_plays.limit(limit) logging.info(f'bat_plays query: {bat_plays}') logging.info(f'run_plays query: {run_plays}') return_stats = { 'count': bat_plays.count(), 'stats': [] } for x in bat_plays: this_run = run_plays.order_by(StratPlay.id) if 'player' in group_by: this_run = this_run.where(StratPlay.runner == x.batter) if 'game' in group_by: this_run = this_run.where(StratPlay.game == x.game) if this_run.count() > 0: sum_sb = this_run[0].sum_sb sum_cs = this_run[0].sum_cs run_wpa = this_run[0].sum_wpa else: sum_sb = 0 sum_cs = 0 run_wpa = 0 this_wpa = bat_plays.where( (StratPlay.wpa >= min_wpa) & (StratPlay.wpa <= max_wpa) & (StratPlay.batter == x.batter) ) if this_wpa.count() > 0: sum_wpa = this_wpa[0].sum_wpa else: sum_wpa = 0 tot_ab = x.sum_ab if x.sum_ab > 0 else 1 obp = (x.sum_hit + x.sum_bb + x.sum_hbp + x.sum_ibb) / x.sum_pa slg = (x.sum_hr * 4 + x.sum_triple * 3 + x.sum_double * 2 + (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr)) / tot_ab this_game = 'TOT' if group_by in ['playergame', 'teamgame']: this_game = x.game_id if short_output else model_to_dict(x.game, recurse=False) lob_all_rate, lob_2outs_rate, rbi_rate = 0, 0, 0 if x.count_runner1 + x.count_runner2 + x.count_runner3 > 0: lob_all_rate = (x.count_lo1 + x.count_lo2 + x.count_lo3) / \ (x.count_runner1 + x.count_runner2 + x.count_runner3) rbi_rate = (x.sum_rbi - x.sum_hr) / (x.count_runner1 + x.count_runner2 + x.count_runner3) return_stats['stats'].append({ 'player': x.batter_id if short_output else model_to_dict(x.batter, recurse=False), 'team': x.batter_team_id if short_output else model_to_dict(x.batter_team, recurse=False), 'pa': x.sum_pa, 'ab': x.sum_ab, 'run': x.sum_run, 'hit': x.sum_hit, 'rbi': x.sum_rbi, 'double': x.sum_double, 'triple': x.sum_triple, 'hr': x.sum_hr, 'bb': x.sum_bb, 'so': x.sum_so, 'hbp': x.sum_hbp, 'sac': x.sum_sac, 'ibb': x.sum_ibb, 'gidp': x.sum_gidp, 'sb': sum_sb, 'cs': sum_cs, 'bphr': x.sum_bphr, 'bpfo': x.sum_bpfo, 'bp1b': x.sum_bp1b, 'bplo': x.sum_bplo, 'wpa': sum_wpa + run_wpa, 'avg': x.sum_hit / tot_ab, 'obp': obp, 'slg': slg, 'ops': obp + slg, 'woba': (.69 * x.sum_bb + .72 * x.sum_hbp + .89 * (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) + 1.27 * x.sum_double + 1.62 * x.sum_triple + 2.1 * x.sum_hr) / max(x.sum_pa - x.sum_ibb, 1), 'game': this_game, 'lob_all': x.count_lo1 + x.count_lo2 + x.count_lo3, 'lob_all_rate': lob_all_rate, 'lob_2outs': x.count_lo1_3out + x.count_lo2_3out + x.count_lo3_3out, 'rbi%': rbi_rate }) db.close() return return_stats @router.get('/pitching') async def get_pitching_totals( season: list = Query(default=None), week: list = Query(default=None), s_type: Literal['regular', 'post', 'total', None] = None, player_id: list = Query(default=None), group_by: Literal['team', 'player', 'playerteam', 'playergame', 'teamgame', 'league'] = 'player', min_pa: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), obc: list = Query(default=None), risp: Optional[bool] = None, inning: list = Query(default=None), sort: Optional[str] = None, limit: Optional[int] = None, short_output: Optional[bool] = False): season_games = StratGame.select() if season is not None: season_games = season_games.where(StratGame.season << season) if week is not None and s_type is not None: raise HTTPException(status_code=400, detail=f'Week and s_type parameters cannot be used in the same query') if week is not None: season_games = season_games.where(StratGame.week << week) if s_type is not None: if s_type == 'regular': season_games = season_games.where(StratGame.week <= 18) elif s_type == 'post': season_games = season_games.where(StratGame.week > 18) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) pit_plays = ( StratPlay .select(StratPlay.pitcher, StratPlay.pitcher_team, StratPlay.game, fn.SUM(StratPlay.pa).alias('sum_pa'), fn.SUM(StratPlay.ab).alias('sum_ab'), fn.SUM(StratPlay.run).alias('sum_run'), fn.SUM(StratPlay.hit).alias('sum_hit'), fn.SUM(StratPlay.rbi).alias('sum_rbi'), fn.SUM(StratPlay.double).alias('sum_double'), fn.SUM(StratPlay.triple).alias('sum_triple'), fn.SUM(StratPlay.homerun).alias('sum_hr'), fn.SUM(StratPlay.bb).alias('sum_bb'), fn.SUM(StratPlay.so).alias('sum_so'), fn.SUM(StratPlay.wpa).alias('sum_wpa'), fn.SUM(StratPlay.hbp).alias('sum_hbp'), fn.SUM(StratPlay.sac).alias('sum_sac'), fn.SUM(StratPlay.ibb).alias('sum_ibb'), fn.SUM(StratPlay.gidp).alias('sum_gidp'), fn.SUM(StratPlay.sb).alias('sum_sb'), fn.SUM(StratPlay.cs).alias('sum_cs'), fn.SUM(StratPlay.bphr).alias('sum_bphr'), fn.SUM(StratPlay.bpfo).alias('sum_bpfo'), fn.SUM(StratPlay.bp1b).alias('sum_bp1b'), fn.SUM(StratPlay.bplo).alias('sum_bplo'), fn.SUM(StratPlay.wild_pitch).alias('sum_wp'), fn.SUM(StratPlay.balk).alias('sum_balk'), fn.SUM(StratPlay.outs).alias('sum_outs'), fn.SUM(StratPlay.e_run).alias('sum_erun'), fn.COUNT(StratPlay.on_first_final).filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4)).alias('count_lo1'), fn.COUNT(StratPlay.on_second_final).filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4)).alias('count_lo2'), fn.COUNT(StratPlay.on_third_final).filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4)).alias('count_lo3'), fn.COUNT(StratPlay.on_first).filter(StratPlay.on_first.is_null(False)).alias('count_runner1'), fn.COUNT(StratPlay.on_second).filter(StratPlay.on_second.is_null(False)).alias('count_runner2'), fn.COUNT(StratPlay.on_third).filter(StratPlay.on_third.is_null(False)).alias('count_runner3'), fn.COUNT(StratPlay.on_first_final).filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_lo1_3out'), fn.COUNT(StratPlay.on_second_final).filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_lo2_3out'), fn.COUNT(StratPlay.on_third_final).filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3)).alias('count_lo3_3out')) .where((StratPlay.game << season_games) & (StratPlay.pitcher.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_pa) ) all_dec = ( Decision .select(Decision.pitcher, fn.SUM(Decision.win).alias('sum_win'), fn.SUM(Decision.loss).alias('sum_loss'), fn.SUM(Decision.hold).alias('sum_hold'), fn.SUM(Decision.is_save).alias('sum_save'), fn.SUM(Decision.b_save).alias('sum_b_save'), fn.SUM(Decision.irunners).alias('sum_irunners'), fn.SUM(Decision.irunners_scored).alias('sum_irun_scored'), fn.SUM(Decision.is_start).alias('sum_gs'), fn.COUNT(Decision.game).alias('sum_game')) .where(Decision.game << season_games) ) if player_id is not None: all_players = Player.select().where(Player.id << player_id) pit_plays = pit_plays.where(StratPlay.pitcher << all_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) pit_plays = pit_plays.where(StratPlay.pitcher_team << all_teams) if obc is not None: pit_plays = pit_plays.where(StratPlay.on_base_code << obc) if risp is not None: pit_plays = pit_plays.where(StratPlay.on_base_code << ['100', '101', '110', '111', '010', '011']) if inning is not None: pit_plays = pit_plays.where(StratPlay.inning_num << inning) if group_by is not None: if group_by == 'player': pit_plays = pit_plays.group_by(StratPlay.pitcher) elif group_by == 'team': pit_plays = pit_plays.group_by(StratPlay.pitcher_team) elif group_by == 'playerteam': pit_plays = pit_plays.group_by(StratPlay.pitcher, StratPlay.pitcher_team) elif group_by == 'playergame': pit_plays = pit_plays.group_by(StratPlay.pitcher, StratPlay.game) elif group_by == 'teamgame': pit_plays = pit_plays.group_by(StratPlay.pitcher_team, StratPlay.game) elif group_by == 'league': pit_plays = pit_plays.join(StratGame) pit_plays = pit_plays.group_by(StratPlay.game.season) if sort is not None: if sort == 'player': pit_plays = pit_plays.order_by(StratPlay.pitcher) elif sort == 'team': pit_plays = pit_plays.order_by(StratPlay.pitcher_team) elif sort == 'wpa-desc': pit_plays = pit_plays.order_by(SQL('sum_wpa').asc()) # functions seem reversed since pitcher plays negative elif sort == 'wpa-asc': pit_plays = pit_plays.order_by(SQL('sum_wpa').desc()) elif sort == 'ip-desc': pit_plays = pit_plays.order_by(SQL('sum_outs').desc()) elif sort == 'ip-asc': pit_plays = pit_plays.order_by(SQL('sum_outs').asc()) elif sort == 'game-desc': pit_plays = pit_plays.order_by(SQL('sum_game').desc()) elif sort == 'game-asc': pit_plays = pit_plays.order_by(SQL('sum_game').asc()) elif sort == 'newest': pit_plays = pit_plays.order_by(StratPlay.game_id.desc(), StratPlay.play_num.desc()) elif sort == 'oldest': pit_plays = pit_plays.order_by(StratPlay.game_id, StratPlay.play_num) if limit is not None: if limit < 1: limit = 1 pit_plays = pit_plays.limit(limit) return_stats = { 'count': pit_plays.count(), 'stats': [] } for x in pit_plays: this_dec = all_dec.where(Decision.pitcher == x.pitcher) tot_outs = x.sum_outs if x.sum_outs > 0 else 1 obp = (x.sum_hit + x.sum_bb + x.sum_hbp + x.sum_ibb) / x.sum_pa slg = (x.sum_hr * 4 + x.sum_triple * 3 + x.sum_double * 2 + (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr)) / x.sum_ab tot_bb = 0.1 if x.sum_bb == 0 else x.sum_bb this_game = 'TOT' if group_by in ['playergame', 'teamgame']: this_game = x.game_id if short_output else model_to_dict(x.game, recurse=False) this_dec = all_dec.where((Decision.pitcher == x.pitcher) & (Decision.game == x.game)) lob_all_rate, lob_2outs_rate, rbi_rate = 0, 0, 0 if x.count_runner1 + x.count_runner2 + x.count_runner3 > 0: lob_all_rate = (x.count_lo1 + x.count_lo2 + x.count_lo3) / \ (x.count_runner1 + x.count_runner2 + x.count_runner3) rbi_rate = (x.sum_rbi - x.sum_hr) / (x.count_runner1 + x.count_runner2 + x.count_runner3) return_stats['stats'].append({ 'player': x.pitcher_id if short_output else model_to_dict(x.pitcher, recurse=False), 'team': x.pitcher_team_id if short_output else model_to_dict(x.pitcher_team, recurse=False), 'tbf': x.sum_pa, 'outs': x.sum_outs, 'games': this_dec[0].sum_game, 'gs': this_dec[0].sum_gs, 'win': this_dec[0].sum_win, 'loss': this_dec[0].sum_loss, 'hold': this_dec[0].sum_hold, 'save': this_dec[0].sum_save, 'bsave': this_dec[0].sum_b_save, 'ir': this_dec[0].sum_irunners, 'ir_sc': this_dec[0].sum_irun_scored, 'ab': x.sum_ab, 'run': x.sum_run, 'e_run': x.sum_erun, 'hits': x.sum_hit, 'double': x.sum_double, 'triple': x.sum_triple, 'hr': x.sum_hr, 'bb': x.sum_bb, 'so': x.sum_so, 'hbp': x.sum_hbp, 'sac': x.sum_sac, 'ibb': x.sum_ibb, 'gidp': x.sum_gidp, 'sb': x.sum_sb, 'cs': x.sum_cs, 'bphr': x.sum_bphr, 'bpfo': x.sum_bpfo, 'bp1b': x.sum_bp1b, 'bplo': x.sum_bplo, 'wp': x.sum_wp, 'balk': x.sum_balk, 'wpa': x.sum_wpa * -1, 'era': (x.sum_erun * 27) / tot_outs, 'whip': ((x.sum_bb + x.sum_hit + x.sum_ibb) * 3) / tot_outs, 'avg': x.sum_hit / x.sum_ab, 'obp': obp, 'slg': slg, 'ops': obp + slg, 'woba': (.69 * x.sum_bb + .72 * x.sum_hbp + .89 * (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) + 1.27 * x.sum_double + 1.62 * x.sum_triple + 2.1 * x.sum_hr) / max(x.sum_pa - x.sum_ibb, 1), 'k/9': x.sum_so * 9 / (tot_outs / 3), 'bb/9': x.sum_bb * 9 / (tot_outs / 3), 'k/bb': x.sum_so / tot_bb, 'game': this_game, 'lob_2outs': x.count_lo1_3out + x.count_lo2_3out + x.count_lo3_3out, 'rbi%': rbi_rate }) db.close() return return_stats @router.get('/fielding') async def get_fielding_totals( season: list = Query(default=None), week: list = Query(default=None), s_type: Literal['regular', 'post', 'total', None] = None, position: list = Query(default=None), player_id: list = Query(default=None), group_by: Literal['team', 'player', 'playerteam', 'playerposition', 'teamposition'] = 'player', min_ch: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), sort: Optional[str] = None, limit: Optional[int] = None, short_output: Optional[bool] = False): season_games = StratGame.select() if season is not None: season_games = season_games.where(StratGame.season << season) if week is not None and s_type is not None: raise HTTPException(status_code=400, detail=f'Week and s_type parameters cannot be used in the same query') if week is not None: season_games = season_games.where(StratGame.week << week) if s_type is not None: if s_type == 'regular': season_games = season_games.where(StratGame.week <= 18) elif s_type == 'post': season_games = season_games.where(StratGame.week > 18) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) def_plays = ( StratPlay .select(StratPlay.defender, StratPlay.defender_team, fn.SUM(StratPlay.error).alias('sum_error'), fn.SUM(StratPlay.hit).alias('sum_hit'), fn.SUM(StratPlay.pa).alias('sum_chances'), fn.SUM(StratPlay.wpa).alias('sum_wpa'), StratPlay.check_pos) .where((StratPlay.game << season_games) & (StratPlay.defender.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_ch) ) cat_plays = ( StratPlay .select(StratPlay.catcher, StratPlay.catcher_team, fn.SUM(StratPlay.sb).alias('sum_sb'), fn.SUM(StratPlay.cs).alias('sum_cs'), fn.SUM(StratPlay.wpa).alias('sum_wpa'), fn.SUM(StratPlay.passed_ball).alias('sum_pb')) .where((StratPlay.game << season_games) & (StratPlay.catcher.is_null(False))) ) if player_id is not None: all_players = Player.select().where(Player.id << player_id) def_plays = def_plays.where(StratPlay.defender << all_players) cat_plays = cat_plays.where(StratPlay.catcher << all_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) def_plays = def_plays.where(StratPlay.defender_team << all_teams) cat_plays = cat_plays.where(StratPlay.catcher_team << all_teams) if position is not None: def_plays = def_plays.where(StratPlay.check_pos << position) if group_by is not None: if group_by == 'player': def_plays = def_plays.group_by(StratPlay.defender) cat_plays = cat_plays.group_by(StratPlay.catcher) elif group_by == 'team': def_plays = def_plays.group_by(StratPlay.defender_team) cat_plays = cat_plays.group_by(StratPlay.catcher_team) elif group_by == 'playerteam': def_plays = def_plays.group_by(StratPlay.defender, StratPlay.defender_team) cat_plays = cat_plays.group_by(StratPlay.catcher, StratPlay.catcher_team) elif group_by == 'playerposition': def_plays = def_plays.group_by(StratPlay.defender, StratPlay.check_pos) cat_plays = cat_plays.group_by(StratPlay.catcher) elif group_by == 'teamposition': def_plays = def_plays.group_by(StratPlay.defender_team, StratPlay.check_pos) cat_plays = cat_plays.group_by(StratPlay.catcher_team) if sort is not None: if sort == 'player': def_plays = def_plays.order_by(StratPlay.defender) elif sort == 'team': def_plays = def_plays.order_by(StratPlay.defender_team) elif sort == 'wpa-desc': def_plays = def_plays.order_by(SQL('sum_wpa').asc()) elif sort == 'wpa-asc': def_plays = def_plays.order_by(SQL('sum_wpa').desc()) elif sort == 'ch-desc': def_plays = def_plays.order_by(SQL('sum_chances').desc()) elif sort == 'ch-asc': def_plays = def_plays.order_by(SQL('sum_chances').asc()) if limit is not None: if limit < 1: limit = 1 def_plays = def_plays.limit(limit) logging.info(f'def_plays query: {def_plays}') return_stats = { 'count': def_plays.count(), 'stats': [] } for x in def_plays: logging.info(f'this_play: {x}') this_cat = cat_plays.where(StratPlay.catcher == x.defender) if this_cat.count() > 0: sum_sb = this_cat[0].sum_sb sum_cs = this_cat[0].sum_cs sum_wpa = this_cat[0].sum_wpa sum_pb = this_cat[0].sum_pb else: sum_sb = 0 sum_cs = 0 sum_wpa = 0 sum_pb = 0 this_pos = 'TOT' if group_by in ['playerposition', 'teamposition']: this_pos = x.check_pos this_player = x.defender_id if short_output else model_to_dict(x.defender, recurse=False) if group_by == 'teamposition': this_player = 'TOT' return_stats['stats'].append({ 'player': this_player, 'team': x.defender_team_id if short_output else model_to_dict(x.defender_team, recurse=False), 'pos': this_pos, 'x-ch': x.sum_chances, 'hit': x.sum_hit, 'error': x.sum_error, 'sb-ch': sum_sb + sum_cs, 'sb': sum_sb, 'cs': sum_cs, 'pb': sum_pb, 'wpa': (x.sum_wpa + sum_wpa) * -1, 'wf%': (x.sum_chances - (x.sum_error * .5) - (x.sum_hit * .75)) / x.sum_chances, 'cs%': sum_cs / (sum_sb + sum_cs) if (sum_sb + sum_cs) > 0 else '' }) db.close() return return_stats @router.get('/{play_id}') async def get_one_play(play_id: int): if StratPlay.get_or_none(StratPlay.id == play_id) is None: db.close() raise HTTPException(status_code=404, detail=f'Play ID {play_id} not found') r_play = model_to_dict(StratPlay.get_by_id(play_id)) db.close() return r_play @router.patch('/{play_id}') async def patch_play(play_id: int, new_play: PlayModel, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'patch_play - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') if StratPlay.get_or_none(StratPlay.id == play_id) is None: db.close() raise HTTPException(status_code=404, detail=f'Play ID {play_id} not found') StratPlay.update(**new_play.dict()).where(StratPlay.id == play_id).execute() r_play = model_to_dict(StratPlay.get_by_id(play_id)) db.close() return r_play @router.post('') async def post_plays(p_list: PlayList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'post_plays - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') new_plays = [] this_game = StratGame.get_or_none(StratGame.id == p_list.plays[0].game_id) if this_game is None: raise HTTPException(status_code=404, detail=f'Game ID {p_list.plays[0].game_id} not found') for play in p_list.plays: this_play = play this_play.inning_half = this_play.inning_half.lower() top_half = this_play.inning_half == 'top' if this_play.batter_team_id is None and this_play.batter_id is not None: this_play.batter_team_id = this_game.away_team.id if top_half else this_game.home_team.id if this_play.pitcher_team_id is None: this_play.pitcher_team_id = this_game.home_team.id if top_half else this_game.away_team.id if this_play.catcher_id is not None: this_play.catcher_team_id = this_game.home_team.id if top_half else this_game.away_team.id if this_play.defender_id is not None: this_play.defender_team_id = this_game.home_team.id if top_half else this_game.away_team.id if this_play.runner_id is not None: this_play.runner_team_id = this_game.away_team.id if top_half else this_game.home_team.id if this_play.pa == 0: this_play.batter_final = None new_plays.append(this_play.dict()) with db.atomic(): for batch in chunked(new_plays, 20): StratPlay.insert_many(batch).on_conflict_replace().execute() db.close() return f'Inserted {len(new_plays)} plays' @router.delete('/{play_id}') async def delete_play(play_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'delete_play - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') this_play = StratPlay.get_or_none(StratPlay.id == play_id) if not this_play: db.close() raise HTTPException(status_code=404, detail=f'Play ID {play_id} not found') count = this_play.delete_instance() db.close() if count == 1: return f'Play {play_id} has been deleted' else: raise HTTPException(status_code=500, detail=f'Play {play_id} could not be deleted') @router.delete('/game/{game_id}') async def delete_plays_game(game_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'delete_plays_game - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') this_game = StratGame.get_or_none(StratGame.id == game_id) if not this_game: db.close() raise HTTPException(status_code=404, detail=f'Game ID {game_id} not found') count = StratPlay.delete().where(StratPlay.game == this_game).execute() db.close() if count > 0: return f'Deleted {count} plays matching Game ID {game_id}' else: raise HTTPException(status_code=500, detail=f'No plays matching Game ID {game_id} were deleted')