import logging from typing import Literal, Optional from fastapi import APIRouter, Query from ...db_engine import ( SQL, StratGame, StratPlay, Team, Player, db, fn, model_to_dict, ) from ...dependencies import add_cache_headers, cache_result, handle_db_errors from .common import build_season_games router = APIRouter() logger = logging.getLogger("discord_app") @router.get("/batting") @handle_db_errors @add_cache_headers(max_age=10 * 60) @cache_result(ttl=5 * 60, key_prefix="plays-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), sbaplayer_id: list = Query(default=None), min_wpa: Optional[float] = -999, max_wpa: Optional[float] = 999, group_by: Literal[ "team", "player", "playerteam", "playergame", "teamgame", "league", "playerweek", "teamweek", "sbaplayer", ] = "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] = 200, short_output: Optional[bool] = False, page_num: Optional[int] = 1, week_start: Optional[int] = None, week_end: Optional[int] = None, min_repri: Optional[int] = None, ): season_games = build_season_games( season, week, s_type, week_start, week_end, manager_id ) # Build SELECT fields conditionally based on group_by base_select_fields = [ 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.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.SUM(StratPlay.re24_primary).alias("sum_repri"), 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"), ] # Add player and team fields based on grouping type if group_by in ["player", "playerteam", "playergame", "playerweek"]: base_select_fields.insert(0, StratPlay.batter) # Add batter as first field if group_by == "sbaplayer": base_select_fields.insert(0, Player.sbaplayer) if group_by in ["team", "playerteam", "teamgame", "teamweek"]: base_select_fields.append(StratPlay.batter_team) bat_plays = ( StratPlay.select(*base_select_fields) .where((StratPlay.game << season_games) & (StratPlay.batter.is_null(False))) .having((fn.SUM(StratPlay.pa) >= min_pa)) ) if min_repri is not None: bat_plays = bat_plays.having(fn.SUM(StratPlay.re24_primary) >= min_repri) # Build running plays SELECT fields conditionally run_select_fields = [ 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"), fn.SUM(StratPlay.re24_running).alias("sum_rerun"), ] if group_by in ["player", "playerteam", "playergame", "playerweek"]: run_select_fields.insert(0, StratPlay.runner) # Add runner as first field if group_by == "sbaplayer": run_select_fields.insert(0, Player.sbaplayer) if group_by in ["team", "playerteam", "teamgame", "teamweek"]: run_select_fields.append(StratPlay.runner_team) run_plays = StratPlay.select(*run_select_fields).where( (StratPlay.game << season_games) & (StratPlay.runner.is_null(False)) ) # Build defensive plays SELECT fields conditionally def_select_fields = [ 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"), ] if group_by in ["player", "playerteam", "playergame", "playerweek"]: def_select_fields.insert(0, StratPlay.defender) # Add defender as first field if group_by == "sbaplayer": def_select_fields.insert(0, Player.sbaplayer) if group_by in ["team", "playerteam", "teamgame", "teamweek"]: def_select_fields.append(StratPlay.defender_team) def_plays = StratPlay.select(*def_select_fields).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 sbaplayer_id is not None: sba_players = Player.select().where(Player.sbaplayer_id << sbaplayer_id) bat_plays = bat_plays.where(StratPlay.batter << sba_players) run_plays = run_plays.where(StratPlay.runner << sba_players) def_plays = def_plays.where(StratPlay.defender << sba_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) # Initialize game_select_fields for use in GROUP BY game_select_fields = [] # Add StratPlay.game to SELECT clause for group_by scenarios that need it if group_by in ["playergame", "teamgame"]: # For playergame/teamgame grouping, build appropriate SELECT fields if group_by == "playergame": game_select_fields = [ StratPlay.batter, StratPlay.game, StratPlay.batter_team, ] else: # teamgame game_select_fields = [StratPlay.batter_team, StratPlay.game] game_bat_plays = ( StratPlay.select( *game_select_fields, 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.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.SUM(StratPlay.re24_primary).alias("sum_repri"), 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.batter.is_null(False))) .having((fn.SUM(StratPlay.pa) >= min_pa)) ) # Apply the same filters that were applied to bat_plays if player_id is not None: all_players = Player.select().where(Player.id << player_id) game_bat_plays = game_bat_plays.where(StratPlay.batter << all_players) if sbaplayer_id is not None: sba_players = Player.select().where(Player.sbaplayer_id << sbaplayer_id) game_bat_plays = game_bat_plays.where(StratPlay.batter << sba_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) game_bat_plays = game_bat_plays.where(StratPlay.batter_team << all_teams) if position is not None: game_bat_plays = game_bat_plays.where(StratPlay.batter_pos << position) if obc is not None: game_bat_plays = game_bat_plays.where(StratPlay.on_base_code << obc) if risp is not None: game_bat_plays = game_bat_plays.where( StratPlay.on_base_code << ["100", "101", "110", "111", "010", "011"] ) if inning is not None: game_bat_plays = game_bat_plays.where(StratPlay.inning_num << inning) if min_repri is not None: game_bat_plays = game_bat_plays.having( fn.SUM(StratPlay.re24_primary) >= min_repri ) bat_plays = game_bat_plays 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": if game_select_fields: bat_plays = bat_plays.group_by(*game_select_fields) else: 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": if game_select_fields: bat_plays = bat_plays.group_by(*game_select_fields) else: bat_plays = bat_plays.group_by(StratPlay.batter_team, StratPlay.game) run_plays = run_plays.group_by(StratPlay.runner_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) elif group_by == "playerweek": bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by(StratPlay.batter, StratPlay.game.week) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by(StratPlay.runner, StratPlay.game.week) elif group_by == "teamweek": bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by(StratPlay.batter_team, StratPlay.game.week) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by(StratPlay.runner_team, StratPlay.game.week) elif group_by == "sbaplayer": bat_plays = bat_plays.join( Player, on=(StratPlay.batter == Player.id) ).where(Player.sbaplayer.is_null(False)) bat_plays = bat_plays.group_by(Player.sbaplayer) run_plays = run_plays.join( Player, on=(StratPlay.runner == Player.id) ).where(Player.sbaplayer.is_null(False)) run_plays = run_plays.group_by(Player.sbaplayer) def_plays = def_plays.join( Player, on=(StratPlay.defender == Player.id) ).where(Player.sbaplayer.is_null(False)) def_plays = def_plays.group_by(Player.sbaplayer) 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 == "repri-desc": bat_plays = bat_plays.order_by(SQL("sum_repri").desc()) elif sort == "repri-asc": bat_plays = bat_plays.order_by(SQL("sum_repri").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": # For grouped queries, only sort by fields in GROUP BY clause if group_by in ["playergame", "teamgame"]: # StratPlay.game is in GROUP BY for these cases bat_plays = bat_plays.order_by(StratPlay.game.desc()) run_plays = run_plays.order_by(StratPlay.game.desc()) # For other group_by values, skip game_id/play_num sorting since they're not in GROUP BY elif sort == "oldest": # For grouped queries, only sort by fields in GROUP BY clause if group_by in ["playergame", "teamgame"]: # StratPlay.game is in GROUP BY for these cases bat_plays = bat_plays.order_by(StratPlay.game.asc()) run_plays = run_plays.order_by(StratPlay.game.asc()) # For other group_by values, skip game_id/play_num sorting since they're not in GROUP BY if limit < 1: limit = 1 bat_plays = bat_plays.paginate(page_num, limit) logger.info(f"bat_plays query: {bat_plays}") logger.info(f"run_plays query: {run_plays}") return_stats = {"count": bat_plays.count(), "stats": []} for x in bat_plays: this_run = run_plays if group_by == "player": this_run = this_run.where(StratPlay.runner == x.batter) elif group_by == "team": this_run = this_run.where(StratPlay.batter_team == x.batter_team) elif group_by == "playerteam": this_run = this_run.where( (StratPlay.runner == x.batter) & (StratPlay.batter_team == x.batter_team) ) elif group_by == "playergame": this_run = this_run.where( (StratPlay.runner == x.batter) & (StratPlay.game == x.game) ) elif group_by == "teamgame": this_run = this_run.where( (StratPlay.batter_team == x.batter_team) & (StratPlay.game == x.game) ) elif group_by == "playerweek": this_run = this_run.where( (StratPlay.runner == x.batter) & (StratPlay.game.week == x.game.week) ) elif group_by == "teamweek": this_run = this_run.where( (StratPlay.batter_team == x.batter_team) & (StratPlay.game.week == x.game.week) ) elif group_by == "sbaplayer": this_run = this_run.where(Player.sbaplayer == x.batter.sbaplayer) 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 sum_rerun = this_run[0].sum_rerun else: sum_sb = 0 sum_cs = 0 run_wpa = 0 sum_rerun = 0 if group_by == "sbaplayer": wpa_filter = Player.sbaplayer == x.batter.sbaplayer repri_filter = Player.sbaplayer == x.batter.sbaplayer else: wpa_filter = StratPlay.batter == x.batter repri_filter = StratPlay.batter == x.batter this_wpa = bat_plays.where( (StratPlay.wpa >= min_wpa) & (StratPlay.wpa <= max_wpa) & wpa_filter ) if this_wpa.count() > 0: sum_wpa = this_wpa[0].sum_wpa else: sum_wpa = 0 this_repri = bat_plays.where(repri_filter) if this_wpa.count() > 0: sum_repri = this_repri[0].sum_repri else: sum_repri = 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) ) this_week = "TOT" if group_by in ["playerweek", "teamweek"]: this_week = x.game.week this_player = "TOT" if group_by == "sbaplayer": this_player = ( x.batter.sbaplayer_id if short_output else model_to_dict(x.batter.sbaplayer, recurse=False) ) elif "player" in group_by: this_player = ( x.batter_id if short_output else model_to_dict(x.batter, 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 ) # Handle team field based on grouping - set to 'TOT' when not grouping by team if hasattr(x, "batter_team") and x.batter_team is not None: team_info = ( x.batter_team_id if short_output else model_to_dict(x.batter_team, recurse=False) ) else: team_info = "TOT" return_stats["stats"].append( { "player": this_player, "team": team_info, "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": ( 0.69 * x.sum_bb + 0.72 * x.sum_hbp + 0.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, "week": this_week, "re24_primary": sum_repri, # 're24_running': sum_rerun } ) db.close() return return_stats