Create batter_arsenal_model.py

models/batter_arsenal_model.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pandas as pd
+
+from models.batter_zone_model import normalize_pitch_family
+
+
+def _empty_batter_arsenal_row(player_name: str) -> dict[str, Any]:
+    out: dict[str, Any] = {
+        "player_name": player_name,
+        "arsenal_sample_size": 0,
+    }
+
+    for family in ["fastball", "breaking", "offspeed"]:
+        out[f"hr_prob_{family}"] = None
+        out[f"hit_prob_{family}"] = None
+        out[f"tb2p_prob_{family}"] = None
+        out[f"whiff_prob_{family}"] = None
+        out[f"damage_prob_{family}"] = None
+        out[f"sample_size_{family}"] = 0
+
+    return out
+
+
+def build_batter_arsenal_feature_row(
+    statcast_df: pd.DataFrame,
+    player_name: str,
+) -> dict[str, Any]:
+    if statcast_df is None or statcast_df.empty:
+        return _empty_batter_arsenal_row(player_name)
+
+    if "player_name" not in statcast_df.columns:
+        return _empty_batter_arsenal_row(player_name)
+
+    df = statcast_df[statcast_df["player_name"].astype(str) == str(player_name)].copy()
+    if df.empty:
+        return _empty_batter_arsenal_row(player_name)
+
+    if "pitch_name" in df.columns:
+        pitch_name_series = df["pitch_name"]
+    elif "pitch_type" in df.columns:
+        pitch_name_series = df["pitch_type"]
+    else:
+        pitch_name_series = pd.Series(["unknown"] * len(df), index=df.index)
+
+    df["pitch_family"] = pitch_name_series.apply(normalize_pitch_family)
+
+    description = df.get("description", pd.Series("", index=df.index)).astype(str).str.lower()
+    events = df.get("events", pd.Series("", index=df.index)).astype(str).str.lower()
+    launch_speed = pd.to_numeric(df.get("launch_speed"), errors="coerce")
+
+    df["hit_flag"] = events.isin(["single", "double", "triple", "home_run"]).astype(int)
+    df["hr_flag"] = events.eq("home_run").astype(int)
+    df["tb2p_flag"] = events.isin(["double", "triple", "home_run"]).astype(int)
+    df["whiff_flag"] = description.isin(["swinging_strike", "swinging_strike_blocked"]).astype(int)
+    df["damage_flag"] = ((launch_speed >= 95).fillna(False) | df["hr_flag"].eq(1)).astype(int)
+
+    out = _empty_batter_arsenal_row(player_name)
+    out["arsenal_sample_size"] = int(len(df))
+
+    for family in ["fastball", "breaking", "offspeed"]:
+        family_df = df[df["pitch_family"] == family].copy()
+        if family_df.empty:
+            continue
+
+        out[f"hr_prob_{family}"] = float(family_df["hr_flag"].mean())
+        out[f"hit_prob_{family}"] = float(family_df["hit_flag"].mean())
+        out[f"tb2p_prob_{family}"] = float(family_df["tb2p_flag"].mean())
+        out[f"whiff_prob_{family}"] = float(family_df["whiff_flag"].mean())
+        out[f"damage_prob_{family}"] = float(family_df["damage_flag"].mean())
+        out[f"sample_size_{family}"] = int(len(family_df))
+
+    return out