Update models/batter_zone_model.py

models/batter_zone_model.py

@@ -133,41 +133,13 @@ def _empty_batter_zone_row(player_name: str) -> dict[str, Any]:
 
     return out
 
-
-def build_batter_zone_feature_row(statcast_df: pd.DataFrame, player_name: str) -> dict[str, Any]:
-    if statcast_df.empty or "player_name" not in statcast_df.columns:
-        return _empty_batter_zone_row(player_name)
-
-    player_names = statcast_df["player_name"].astype(str).fillna("")
-    normalized_target_variants = _to_last_first_variants(player_name)
-
-    normalized_series = player_names.map(_normalize_name_text)
-
-    mask = normalized_series.isin(normalized_target_variants)
-
-    df = statcast_df[mask].copy()
-
-    if df.empty:
-        normalized_player_name = _normalize_name_text(player_name)
-        name_parts = normalized_player_name.split()
-
-        if len(name_parts) >= 2:
-            first = name_parts[0]
-            last = name_parts[-1]
-
-            loose_mask = normalized_series.apply(
-                lambda n: isinstance(n, str) and first in n and last in n
-            )
-            df = statcast_df[loose_mask].copy()
-
+def _build_zone_metrics_from_df(df: pd.DataFrame, player_name: str) -> dict[str, Any]:
     if df.empty:
         return _empty_batter_zone_row(player_name)
 
-    # Need pitch location + pitch type for zone modeling
     if "plate_x" not in df.columns or "plate_z" not in df.columns:
         return _empty_batter_zone_row(player_name)
 
-    pitch_name_series = None
     if "pitch_name" in df.columns:
         pitch_name_series = df["pitch_name"]
     elif "pitch_type" in df.columns:
@@ -189,7 +161,6 @@ def build_batter_zone_feature_row(statcast_df: pd.DataFrame, player_name: str) -
     estimated_woba = pd.to_numeric(df.get("estimated_woba_using_speedangle"), errors="coerce")
     events = df.get("events", pd.Series(index=df.index, dtype="object")).astype(str).str.lower()
 
-    # rough hit / tb / hr / whiff proxies
     hit_mask = events.isin({"single", "double", "triple", "home_run"})
     hr_mask = events.eq("home_run")
     tb2p_mask = events.isin({"double", "triple", "home_run"})
@@ -197,7 +168,6 @@ def build_batter_zone_feature_row(statcast_df: pd.DataFrame, player_name: str) -
     description_series = df.get("description", pd.Series(index=df.index, dtype="object")).astype(str).str.lower()
     whiff_mask = description_series.isin({"swinging_strike", "swinging_strike_blocked"})
 
-    # damage proxy: either quality contact or strong xwOBA
     damage_mask = (
         (launch_speed >= 95)
         | (estimated_woba >= 0.500)
@@ -214,14 +184,121 @@ def build_batter_zone_feature_row(statcast_df: pd.DataFrame, player_name: str) -
                 continue
 
             subset_idx = subset.index
-
             sample_size = int(len(subset))
             out[f"sample_size_{family}_{zone}"] = sample_size
-
             out[f"hit_prob_{family}_{zone}"] = float(hit_mask.loc[subset_idx].mean())
             out[f"hr_prob_{family}_{zone}"] = float(hr_mask.loc[subset_idx].mean())
             out[f"tb2p_prob_{family}_{zone}"] = float(tb2p_mask.loc[subset_idx].mean())
             out[f"whiff_prob_{family}_{zone}"] = float(whiff_mask.loc[subset_idx].mean())
             out[f"damage_prob_{family}_{zone}"] = float(damage_mask.loc[subset_idx].mean())
 
+    return out
+
+def _blend_rate_with_sample_size(
+    current_value: Any,
+    current_sample: int,
+    previous_value: Any,
+    previous_sample: int,
+) -> float | None:
+    try:
+        current_sample = int(current_sample or 0)
+    except Exception:
+        current_sample = 0
+
+    try:
+        previous_sample = int(previous_sample or 0)
+    except Exception:
+        previous_sample = 0
+
+    current_exists = current_value is not None
+    previous_exists = previous_value is not None
+
+    if not current_exists and not previous_exists:
+        return None
+    if current_exists and not previous_exists:
+        return float(current_value)
+    if previous_exists and not current_exists:
+        return float(previous_value)
+
+    # Current season weight rises with sample size, but does not go to 100% too quickly.
+    current_weight = min(0.85, max(0.25, current_sample / (current_sample + 150.0)))
+    previous_weight = 1.0 - current_weight
+
+    return (float(current_value) * current_weight) + (float(previous_value) * previous_weight)
+
+def build_batter_zone_feature_row(
+    current_statcast_df: pd.DataFrame,
+    player_name: str,
+    previous_statcast_df: pd.DataFrame | None = None,
+) -> dict[str, Any]:
+    if current_statcast_df is None:
+        current_statcast_df = pd.DataFrame()
+    if previous_statcast_df is None:
+        previous_statcast_df = pd.DataFrame()
+
+    def _match_player(df: pd.DataFrame, player_name: str) -> pd.DataFrame:
+        if df.empty or "player_name" not in df.columns:
+            return pd.DataFrame()
+
+        player_names = df["player_name"].astype(str).fillna("")
+        normalized_target_variants = _to_last_first_variants(player_name)
+        normalized_series = player_names.map(_normalize_name_text)
+
+        mask = normalized_series.isin(normalized_target_variants)
+        matched = df[mask].copy()
+
+        if matched.empty:
+            normalized_player_name = _normalize_name_text(player_name)
+            name_parts = normalized_player_name.split()
+
+            if len(name_parts) >= 2:
+                first = name_parts[0]
+                last = name_parts[-1]
+
+                loose_mask = normalized_series.apply(
+                    lambda n: isinstance(n, str) and first in n and last in n
+                )
+                matched = df[loose_mask].copy()
+
+        return matched
+
+    current_df = _match_player(current_statcast_df, player_name)
+    previous_df = _match_player(previous_statcast_df, player_name)
+
+    current_metrics = _build_zone_metrics_from_df(current_df, player_name)
+    previous_metrics = _build_zone_metrics_from_df(previous_df, player_name)
+
+    out = _empty_batter_zone_row(player_name)
+
+    out["zone_sample_size"] = int(current_metrics.get("zone_sample_size", 0) or 0) + int(
+        previous_metrics.get("zone_sample_size", 0) or 0
+    )
+
+    for family in ["fastball", "breaking", "offspeed"]:
+        for zone in ["heart", "shadow", "chase", "waste"]:
+            sample_key = f"sample_size_{family}_{zone}"
+
+            current_sample = int(current_metrics.get(sample_key, 0) or 0)
+            previous_sample = int(previous_metrics.get(sample_key, 0) or 0)
+
+            out[sample_key] = current_sample + previous_sample
+
+            for metric_prefix in [
+                "hit_prob",
+                "hr_prob",
+                "tb2p_prob",
+                "whiff_prob",
+                "damage_prob",
+            ]:
+                key = f"{metric_prefix}_{family}_{zone}"
+                out[key] = _blend_rate_with_sample_size(
+                    current_value=current_metrics.get(key),
+                    current_sample=current_sample,
+                    previous_value=previous_metrics.get(key),
+                    previous_sample=previous_sample,
+                )
+
+    out["current_zone_sample_size"] = int(current_metrics.get("zone_sample_size", 0) or 0)
+    out["previous_zone_sample_size"] = int(previous_metrics.get("zone_sample_size", 0) or 0)
+
     return out