Card Lab Pass 5: identity map + batter_id-based hitter query

- Build player_identity_map.parquet from pybaseball + MLBAM lookup + DB enrichment
- Pitcher statcast_name resolved from ec.player_name (Last, First format)
- Hitter statcast_name set to canonical_name; hitter cards now query by ec.batter = :batter_id
- card_lab_pitcher_seasons.parquet and card_lab_hitter_seasons.parquet built from
pybaseball pitching_stats/batting_stats with IP>0/AB>0 gates
- get_player_card_window_df: add batter_id param for Hitter path (ec.batter = :batter_id)
- card_lab_page.py: read batter_id from Parquet, pass to window query; no DB call for MLBAM ID
- Collision-safe display_names; all validation checks pass (ALL PASS)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

data/card_lab_hitter_seasons.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a661d5caf2cf95decb7a632bc33c0eb33d3268ec689d7bbcccd46abb7c2f0885
+size 92044

data/card_lab_pitcher_seasons.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:7326980b9dffdbc208712b87faf8ed00706aea6172b5ad9e50b9250a84625308
+size 102665

data/player_identity_map.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6036880b0335b6e28010c0540aebb65af0b3fe8946f526ad2d604b08aff4a842
+size 206860

scripts/build_card_lab_season_summaries.py

@@ -240,12 +240,12 @@ def build_season_summaries(seasons: list[int]) -> None:
     pitcher_df = build_pitcher_seasons(seasons, identity_df)
     _PITCHER_OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
     pitcher_df.to_parquet(_PITCHER_OUTPUT_PATH, index=False)
-    print(f"\nWrote {len(pitcher_df)} pitcher rows → {_PITCHER_OUTPUT_PATH}")
+    print(f"\nWrote {len(pitcher_df)} pitcher rows -> {_PITCHER_OUTPUT_PATH}")
 
     hitter_df = build_hitter_seasons(seasons, identity_df)
     _HITTER_OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
     hitter_df.to_parquet(_HITTER_OUTPUT_PATH, index=False)
-    print(f"\nWrote {len(hitter_df)} hitter rows → {_HITTER_OUTPUT_PATH}")
+    print(f"\nWrote {len(hitter_df)} hitter rows -> {_HITTER_OUTPUT_PATH}")
 
 
 if __name__ == "__main__":

scripts/build_player_identity_map.py

@@ -170,11 +170,19 @@ def _build_raw_records(
 
 def _enrich_statcast_names(identity_df: pd.DataFrame) -> pd.DataFrame:
     """
-    Populate statcast_name by matching against statcast_event_core.player_name.
-    This is REQUIRED — without it, hitter rows stay statcast_name=None and are
-    excluded from the hitter selector at build time.
+    Populate statcast_name for each identity record.
+
+    DB context: statcast_event_core.player_name stores the PITCHER name in "Last, First"
+    format.  Pure hitters never appear there.  Strategy:
+
+    - role_pitcher records: match against ec.player_name (Last, First → canonical → norm).
+      statcast_name = the DB's "Last, First" string (used in pitcher selector label only;
+      pitcher window queries use ec.pitcher = :pitcher_id, not player_name).
+    - role_hitter-only records: set statcast_name = canonical_name (First Last).
+      Hitter window queries will use ec.batter = :batter_id via player_id, not player_name.
+    - Two-way players (both roles): try pitcher match first; fall back to canonical_name.
     """
-    print("  Connecting to DB for statcast_name enrichment...")
+    print("  Connecting to DB for statcast_name enrichment (pitchers only)...")
     conn = get_connection()
     try:
         rows = conn.execute(
@@ -184,47 +192,74 @@ def _enrich_statcast_names(identity_df: pd.DataFrame) -> pd.DataFrame:
         conn.close()
 
     statcast_names: list[str] = [r[0] for r in rows if r[0]]
-    statcast_names_set: set[str] = set(statcast_names)
-    print(f"  Loaded {len(statcast_names)} distinct statcast player_names")
+    print(f"  Loaded {len(statcast_names)} distinct statcast player_names (pitcher names)")
 
-    # Build normalized key → [statcast_names] map for fallback matching
+    # Pre-build O(1) lookup dicts so the per-record loop is fast.
+    # statcast names are "Last, First" — apply to_canonical_name before normalizing
+    # so the key matches pybaseball "First Last" canonical_name_normalized.
+    canonical_lower_to_statcast: dict[str, str] = {}   # "first last" → "Last, First"
     norm_to_statcast: dict[str, list[str]] = defaultdict(list)
     for sc in statcast_names:
-        norm_to_statcast[normalize_for_matching(sc)].append(sc)
+        canon = to_canonical_name(sc)                          # "Last, First" → "First Last"
+        cl = canon.lower()
+        if cl not in canonical_lower_to_statcast:
+            canonical_lower_to_statcast[cl] = sc
+        norm_to_statcast[normalize_for_matching(canon)].append(sc)
 
-    resolved = 0
-    ambiguous = 0
-    unmatched = 0
+    resolved_pitcher = 0
+    resolved_hitter  = 0
+    ambiguous        = 0
+    unmatched_pitcher = 0
 
     for idx, row in identity_df.iterrows():
         canonical = row["canonical_name"]
         norm_key  = row["canonical_name_normalized"]
-
-        # Layer 1: exact match (case-insensitive)
-        sc_lower = canonical.lower()
-        exact = [sc for sc in statcast_names_set if sc.lower() == sc_lower]
-        if len(exact) == 1:
-            identity_df.at[idx, "statcast_name"] = exact[0]
-            if "statcast" not in str(row.get("source_note", "")):
+        is_pitcher = bool(row.get("role_pitcher"))
+        is_hitter  = bool(row.get("role_hitter"))
+
+        if is_pitcher:
+            # Layer 1: exact canonical lowercase match (O(1))
+            sc = canonical_lower_to_statcast.get(canonical.lower())
+            if sc:
+                identity_df.at[idx, "statcast_name"] = sc
                 identity_df.at[idx, "source_note"] = str(row.get("source_note", "")) + "+statcast"
-            resolved += 1
-            continue
-
-        # Layer 2: normalized key match
-        candidates = norm_to_statcast.get(norm_key, [])
-        if len(candidates) == 1:
-            identity_df.at[idx, "statcast_name"] = candidates[0]
-            if "statcast" not in str(row.get("source_note", "")):
-                identity_df.at[idx, "source_note"] = str(row.get("source_note", "")) + "+statcast-norm"
-            resolved += 1
-        elif len(candidates) > 1:
-            print(f"  AMBIGUOUS: {canonical!r} → {candidates}")
-            ambiguous += 1
-        else:
-            print(f"  UNMATCHED: {canonical!r} (no statcast player_name found)")
-            unmatched += 1
+                resolved_pitcher += 1
+                continue
 
-    print(f"  Enrichment: resolved={resolved} ambiguous={ambiguous} unmatched={unmatched}")
+            # Layer 2: normalized key match (O(1))
+            candidates = norm_to_statcast.get(norm_key, [])
+            if len(candidates) == 1:
+                identity_df.at[idx, "statcast_name"] = candidates[0]
+                identity_df.at[idx, "source_note"] = str(row.get("source_note", "")) + "+statcast-norm"
+                resolved_pitcher += 1
+                continue
+            elif len(candidates) > 1:
+                print(f"  AMBIGUOUS pitcher: {canonical!r} → {candidates}")
+                ambiguous += 1
+                # Fall through: use canonical_name as statcast_name so row is not excluded
+            else:
+                print(f"  UNMATCHED pitcher: {canonical!r}")
+                unmatched_pitcher += 1
+                # Fall through to hitter branch if also a hitter; else canonical fallback
+
+        if is_hitter and identity_df.at[idx, "statcast_name"] is None:
+            # Hitter window queries use ec.batter = :batter_id (player_id), not player_name.
+            # statcast_name must be non-null to pass build validation.
+            # Use canonical_name (First Last) as a stable non-null placeholder.
+            identity_df.at[idx, "statcast_name"] = canonical
+            identity_df.at[idx, "source_note"] = str(row.get("source_note", "")) + "+hitter-canonical"
+            resolved_hitter += 1
+
+        # Final fallback: any record still missing statcast_name (e.g. unmatched pure pitcher)
+        if identity_df.at[idx, "statcast_name"] is None:
+            identity_df.at[idx, "statcast_name"] = canonical
+            identity_df.at[idx, "source_note"] = str(row.get("source_note", "")) + "+canonical-fallback"
+
+    print(
+        f"  Enrichment: pitcher_matched={resolved_pitcher} "
+        f"hitter_canonical={resolved_hitter} "
+        f"ambiguous={ambiguous} unmatched_pitcher={unmatched_pitcher}"
+    )
     return identity_df
 
 
@@ -242,31 +277,29 @@ def _resolve_collisions(identity_df: pd.DataFrame) -> pd.DataFrame:
     identity_df = identity_df.copy()
     identity_df["display_name"] = identity_df["canonical_name"]
 
-    # Find collisions: same normalized name, different player_id
+    # Find collisions: same normalized name, multiple rows
     norm_groups = identity_df.groupby("canonical_name_normalized")
     collision_count = 0
     for norm_key, group in norm_groups:
-        # Only a collision if multiple distinct player_ids exist in the group
-        ids = group["player_id"].dropna().unique()
-        if len(ids) <= 1 and len(group) == 1:
-            continue
-        if len(ids) > 1:
-            # Sort by player_id ascending (deterministic) — None last
-            sorted_group = group.sort_values(
-                "player_id", ascending=True, na_position="last"
-            )
-            for rank, (idx, row) in enumerate(sorted_group.iterrows()):
-                if rank == 0:
-                    # Primary: keep canonical_name as display_name
-                    pass
-                else:
-                    pid = row.get("player_id")
-                    suffix = str(int(pid)) if pd.notna(pid) else "?"
-                    identity_df.at[idx, "display_name"] = f"{row['canonical_name']} ({suffix})"
-                    identity_df.at[idx, "source_note"] = (
-                        str(row.get("source_note", "")) + "+collision-resolved"
-                    )
-                    collision_count += 1
+        if len(group) == 1:
+            continue  # no collision
+        # Multiple records share the same normalized name — assign suffixes to all but the first.
+        # Sort: non-null player_id ascending first, then null (deterministic).
+        sorted_group = group.sort_values(
+            "player_id", ascending=True, na_position="last"
+        )
+        for rank, (idx, row) in enumerate(sorted_group.iterrows()):
+            if rank == 0:
+                # Primary: keep canonical_name as display_name
+                pass
+            else:
+                pid = row.get("player_id")
+                suffix = str(int(pid)) if pd.notna(pid) else "?"
+                identity_df.at[idx, "display_name"] = f"{row['canonical_name']} ({suffix})"
+                identity_df.at[idx, "source_note"] = (
+                    str(row.get("source_note", "")) + "+collision-resolved"
+                )
+                collision_count += 1
 
     if collision_count:
         print(f"  Resolved {collision_count} collision suffix(es)")
@@ -309,7 +342,7 @@ def build_identity_map(seasons: list[int]) -> None:
 
     OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
     identity_df.to_parquet(OUTPUT_PATH, index=False)
-    print(f"\nWrote {len(identity_df)} rows → {OUTPUT_PATH}")
+    print(f"\nWrote {len(identity_df)} rows -> {OUTPUT_PATH}")
 
     # Summary
     with_mlbam   = identity_df["player_id"].notna().sum()

visualization/card_lab_page.py

@@ -47,10 +47,11 @@ def normalize_name(name: str) -> str:
 # Card generation functions — button-click only, no caching
 # ---------------------------------------------------------------------------
 
-def _gen_hitter_bytes(conn, player_name, mode, year, date, start_date, end_date, fmt, player_pil):
+def _gen_hitter_bytes(conn, player_name, batter_id, mode, year, date, start_date, end_date, fmt, player_pil):
     windowed_df = get_player_card_window_df(
         conn, player_name, "Hitter", mode=mode, year=year,
         date=date, start_date=start_date, end_date=end_date,
+        batter_id=batter_id,
     )
     if windowed_df.empty:
         return None, "", "limited"
@@ -149,6 +150,7 @@ def render_card_lab(conn) -> None:
     # Hitter pipeline variables
     hitter_display_names:       list[str]                     = []
     hitter_display_to_statcast: dict[str, str]                = {}
+    _hitter_batter_id:          int | None                    = None   # MLBAM batter id from Parquet
 
     # Pitcher pipeline variables
     pitcher_display_names:      list[str]                     = []
@@ -169,15 +171,31 @@ def render_card_lab(conn) -> None:
             return
 
         selected_hitter_display = st.selectbox("Player", hitter_display_names, key="cl_player_hitter")
-        # statcast_name is required — null rows excluded at build time, no runtime fallback allowed
-        statcast_name = hitter_display_to_statcast.get(selected_hitter_display)
-        if not statcast_name:
-            raise RuntimeError(
-                f"Missing statcast_name for hitter selector row: {selected_hitter_display!r}. "
-                "Re-run scripts/build_player_identity_map.py with DB enrichment."
+
+        # Read batter_id (MLBAM) from the Parquet — used for ec.batter = :batter_id query.
+        # DB schema: ec.player_name = PITCHER, not batter.  batter_id is required.
+        _hitter_batter_id: int | None = None
+        if _HITTER_SEASONS_PATH.exists():
+            import pandas as _pd_hi
+            _hdf = _pd_hi.read_parquet(_HITTER_SEASONS_PATH)
+            _match = _hdf[_hdf["display_name"] == selected_hitter_display]
+            if selector_year is not None:
+                _yr = _match[_match["Season"] == selector_year]
+                if not _yr.empty:
+                    _match = _yr
+            if not _match.empty:
+                _pid = _match.iloc[0]["player_id"]
+                if _pd_hi.notna(_pid):
+                    _hitter_batter_id = int(_pid)
+
+        if _hitter_batter_id is None:
+            st.warning(
+                f"No MLBAM batter ID found for {selected_hitter_display!r}. "
+                "Card data will be empty. Re-run build_player_identity_map.py."
             )
-        player_name         = statcast_name             # exact name for DB queries
-        player_name_display = selected_hitter_display   # display-only
+
+        player_name         = selected_hitter_display   # display label (injected as literal in SQL)
+        player_name_display = selected_hitter_display
 
     elif card_type == "Pitcher":
         # Source: data/card_lab_pitcher_seasons.parquet (pybaseball pitching_stats, IP > 0 gate)
@@ -239,7 +257,8 @@ def render_card_lab(conn) -> None:
     mlbam_id: int | None = None
     player_pil = None
     if card_type == "Hitter" and player_name:
-        mlbam_id = get_batter_mlbam_id(conn, player_name, selector_year)
+        # batter_id from Parquet — no DB call needed (ec.player_name = pitcher, not batter)
+        mlbam_id = _hitter_batter_id
         player_pil = resolve_player_image(mlbam_id) if mlbam_id else None
     elif card_type == "Pitcher" and pitcher_id:
         mlbam_id = pitcher_id
@@ -300,6 +319,8 @@ def render_card_lab(conn) -> None:
 
         if card_type in ("Hitter", "Pitcher"):
             st.write(f"**MLBAM ID:** {mlbam_id}")
+            if card_type == "Hitter":
+                st.write(f"**Batter ID (from Parquet):** {_hitter_batter_id}")
             st.write(f"**Image fetched:** {player_pil is not None}")
 
     fmt = st.radio("Format", ["PNG", "JPG"], horizontal=True, key="cl_fmt")
@@ -321,7 +342,7 @@ def render_card_lab(conn) -> None:
                 status.info("Querying warehouse data...")
                 status.info("Building poster...")
                 img_bytes, tf, dq = _gen_hitter_bytes(
-                    conn, player_name, mode_key, year, date, start_date, end_date, fmt, player_pil
+                    conn, player_name, _hitter_batter_id, mode_key, year, date, start_date, end_date, fmt, player_pil
                 )
                 st.session_state["card_player"]    = normalize_name(player_name or "unknown")
                 st.session_state["card_timeframe"] = tf

visualization/cards/card_queries.py

@@ -228,6 +228,7 @@ def get_player_card_window_df(
     start_date: str | None = None,
     end_date: str | None = None,
     pitcher_id: int | None = None,
+    batter_id: int | None = None,
 ) -> pd.DataFrame:
     """
     Fetch the FULL matching pitch/event dataset for one player over a time window.
@@ -238,8 +239,9 @@ def get_player_card_window_df(
     Season mode uses source_season INT filter (fastest).
     Date range / single date use game_date TEXT range comparison.
 
-    For Pitcher card_type, pitcher_id is REQUIRED (the numeric ID from statcast_event_core).
-    The player_name string is injected as a display label for the feature builders.
+    For Pitcher card_type, pitcher_id is REQUIRED (ec.pitcher = :pitcher_id).
+    For Hitter card_type, batter_id is REQUIRED (ec.batter = :batter_id).
+    player_name is injected as a display label in both cases.
     """
     try:
         if card_type == "Pitcher":
@@ -289,14 +291,54 @@ def get_player_card_window_df(
                 params = {"player_name": player_name, "pitcher_id": pitcher_id, "sd": sd, "ed": ed}
 
         else:  # Hitter
-            _HITTER_WHERE = f"SELECT {_HITTER_JOIN_SELECT} {_HITTER_JOIN_FROM} WHERE ec.player_name = :player_name"
+            if batter_id is None:
+                logger.warning(
+                    "[card_lab_db_window] batter_id required for Hitter card_type "
+                    "(player_name='%s') — returning empty", player_name,
+                )
+                return pd.DataFrame()
+
+            # DB schema: ec.player_name = PITCHER name.  Hitters are identified by
+            # ec.batter (MLBAM ID).  Inject player_name as a literal so downstream
+            # card_data.py filtering (df["player_name"] == player_name) still works.
+            _HITTER_SELECT = f"""
+                SELECT
+                    ec.event_key,
+                    :player_name   AS player_name,
+                    ec.game_date,
+                    ec.game_pk,
+                    ec.source_season,
+                    ec.pitch_name,
+                    ec.events,
+                    ec.description,
+                    ec.stand,
+                    ec.p_throws,
+                    ec.batter,
+                    ec.home_team,
+                    ec.away_team,
+                    ec.inning,
+                    ec.at_bat_number,
+                    ec.pitch_number,
+                    bb.launch_speed,
+                    bb.launch_angle,
+                    bb.bb_type,
+                    bb.estimated_woba_using_speedangle,
+                    pr.release_speed,
+                    pr.release_spin_rate,
+                    pr.pfx_x,
+                    pr.pfx_z
+                FROM statcast_event_core ec
+                LEFT JOIN statcast_batted_ball   bb ON ec.event_key = bb.event_key
+                LEFT JOIN statcast_pitch_release pr ON ec.event_key = pr.event_key
+                WHERE ec.batter = :batter_id
+            """
             if mode == "season" and year:
-                sql = text(_HITTER_WHERE + " AND ec.source_season = :year ORDER BY ec.game_date DESC, ec.game_pk DESC")
-                params = {"player_name": player_name, "year": int(year)}
+                sql = text(_HITTER_SELECT + " AND ec.source_season = :year ORDER BY ec.game_date DESC, ec.game_pk DESC")
+                params = {"player_name": player_name, "batter_id": batter_id, "year": int(year)}
             else:
                 sd, ed = _date_range(mode, date, start_date, end_date)
-                sql = text(_HITTER_WHERE + " AND ec.game_date >= :sd AND ec.game_date <= :ed ORDER BY ec.game_date DESC, ec.game_pk DESC")
-                params = {"player_name": player_name, "sd": sd, "ed": ed}
+                sql = text(_HITTER_SELECT + " AND ec.game_date >= :sd AND ec.game_date <= :ed ORDER BY ec.game_date DESC, ec.game_pk DESC")
+                params = {"player_name": player_name, "batter_id": batter_id, "sd": sd, "ed": ed}
 
         df = pd.read_sql(sql, conn, params=params)