Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| from typing import Any | |
| def _safe_float(value: Any) -> float | None: | |
| try: | |
| if value is None: | |
| return None | |
| text = str(value).strip().lower() | |
| if text in {"", "nan", "none"}: | |
| return None | |
| return float(value) | |
| except Exception: | |
| return None | |
| def calculate_matchup_score( | |
| batter_row: Any = None, | |
| pitcher_profile: dict[str, Any] | None = None, | |
| venue_name: str | None = None, | |
| temperature_f: Any = None, | |
| wind_speed_mph: Any = None, | |
| pitcher_adj: dict[str, Any] | None = None, | |
| zone_matchup_adj: dict[str, Any] | None = None, | |
| **kwargs, | |
| ) -> dict[str, Any]: | |
| pitcher_profile = pitcher_profile or {} | |
| pitcher_adj = pitcher_adj or {} | |
| zone_matchup_adj = zone_matchup_adj or {} | |
| # If richer live-state pitcher adjustments are not passed, fall back to pitcher_profile. | |
| if not pitcher_adj: | |
| pitcher_adj = pitcher_profile | |
| notes: list[str] = [] | |
| matchup_score = 50.0 | |
| # ---------------------------- | |
| # Batter quality | |
| # ---------------------------- | |
| ev90 = _safe_float(getattr(batter_row, "get", lambda *_: None)("ev90")) | |
| hard_hit_rate = _safe_float(getattr(batter_row, "get", lambda *_: None)("hard_hit_rate")) | |
| barrel_rate = _safe_float(getattr(batter_row, "get", lambda *_: None)("barrel_rate")) | |
| xwoba = _safe_float(getattr(batter_row, "get", lambda *_: None)("xwoba")) | |
| if ev90 is not None: | |
| ev_component = max(-8.0, min(10.0, (ev90 - 100.0) * 1.5)) | |
| matchup_score += ev_component | |
| if ev_component >= 4: | |
| notes.append("Strong EV90") | |
| if hard_hit_rate is not None: | |
| hh_component = max(-5.0, min(6.0, (hard_hit_rate - 0.40) * 20.0)) | |
| matchup_score += hh_component | |
| if hh_component >= 2: | |
| notes.append("Hard-hit edge") | |
| if barrel_rate is not None: | |
| barrel_component = max(-4.0, min(7.0, (barrel_rate - 0.07) * 45.0)) | |
| matchup_score += barrel_component | |
| if barrel_component >= 2: | |
| notes.append("Barrel upside") | |
| if xwoba is not None: | |
| xwoba_component = max(-5.0, min(7.0, (xwoba - 0.320) * 35.0)) | |
| matchup_score += xwoba_component | |
| if xwoba_component >= 2: | |
| notes.append("xwOBA support") | |
| # ---------------------------- | |
| # Pitcher weakness / live weakness | |
| # ---------------------------- | |
| fatigue_score = _safe_float(pitcher_adj.get("fatigue_score")) | |
| degradation_score = _safe_float(pitcher_adj.get("degradation_score")) | |
| velo_delta = _safe_float(pitcher_adj.get("velo_delta")) | |
| spin_delta = _safe_float(pitcher_adj.get("spin_delta")) | |
| extension_delta = _safe_float(pitcher_adj.get("extension_delta")) | |
| pitch_count = _safe_float(pitcher_adj.get("pitch_count")) | |
| times_through_order = _safe_float(pitcher_adj.get("times_through_order")) | |
| trust_live_score = _safe_float(pitcher_adj.get("trust_live_score")) | |
| # fallback to pitcher_profile if those fields live there instead | |
| if fatigue_score is None: | |
| fatigue_score = _safe_float(pitcher_profile.get("fatigue_score")) | |
| if degradation_score is None: | |
| degradation_score = _safe_float(pitcher_profile.get("degradation_score")) | |
| if velo_delta is None: | |
| velo_delta = _safe_float(pitcher_profile.get("velo_delta")) | |
| if spin_delta is None: | |
| spin_delta = _safe_float(pitcher_profile.get("spin_delta")) | |
| if extension_delta is None: | |
| extension_delta = _safe_float(pitcher_profile.get("extension_delta")) | |
| if pitch_count is None: | |
| pitch_count = _safe_float(pitcher_profile.get("pitch_count")) | |
| if times_through_order is None: | |
| times_through_order = _safe_float(pitcher_profile.get("times_through_order")) | |
| if trust_live_score is None: | |
| trust_live_score = _safe_float(pitcher_profile.get("trust_live_score")) | |
| # older/static pitcher profile fields | |
| ev_allowed = _safe_float(pitcher_profile.get("ev_allowed")) | |
| hard_hit_rate_allowed = _safe_float(pitcher_profile.get("hard_hit_rate_allowed")) | |
| barrel_rate_allowed = _safe_float(pitcher_profile.get("barrel_rate_allowed")) | |
| hr_per_pa = _safe_float(pitcher_profile.get("hr_per_pa")) | |
| xwoba_allowed = _safe_float(pitcher_profile.get("xwoba_allowed")) | |
| if ev_allowed is not None: | |
| ev_allowed_component = max(-4.0, min(6.0, (ev_allowed - 89.0) * 1.2)) | |
| matchup_score += ev_allowed_component | |
| if ev_allowed_component >= 2: | |
| notes.append("Pitcher EV allowed") | |
| if hard_hit_rate_allowed is not None: | |
| hh_allowed_component = max(-3.0, min(5.0, (hard_hit_rate_allowed - 0.38) * 16.0)) | |
| matchup_score += hh_allowed_component | |
| if hh_allowed_component >= 1.5: | |
| notes.append("Hard contact allowed") | |
| if barrel_rate_allowed is not None: | |
| barrel_allowed_component = max(-3.0, min(6.0, (barrel_rate_allowed - 0.07) * 35.0)) | |
| matchup_score += barrel_allowed_component | |
| if barrel_allowed_component >= 1.5: | |
| notes.append("Barrels allowed") | |
| if hr_per_pa is not None: | |
| hr_allowed_component = max(-3.0, min(6.0, (hr_per_pa - 0.04) * 60.0)) | |
| matchup_score += hr_allowed_component | |
| if hr_allowed_component >= 1.5: | |
| notes.append("HR-prone pitcher") | |
| if xwoba_allowed is not None: | |
| xwoba_allowed_component = max(-3.0, min(5.0, (xwoba_allowed - 0.315) * 28.0)) | |
| matchup_score += xwoba_allowed_component | |
| if xwoba_allowed_component >= 1.5: | |
| notes.append("xwOBA allowed") | |
| if fatigue_score is not None: | |
| fatigue_component = max(-2.0, min(8.0, fatigue_score * 8.0)) | |
| matchup_score += fatigue_component | |
| if fatigue_component >= 2: | |
| notes.append("Pitcher fatigue") | |
| if degradation_score is not None: | |
| degradation_component = max(-2.0, min(10.0, degradation_score * 10.0)) | |
| matchup_score += degradation_component | |
| if degradation_component >= 2: | |
| notes.append("Live degradation") | |
| if velo_delta is not None: | |
| velo_component = max(-2.0, min(6.0, (-velo_delta) * 3.0)) | |
| matchup_score += velo_component | |
| if velo_component >= 1.5: | |
| notes.append("Velo down") | |
| if spin_delta is not None: | |
| spin_component = max(-2.0, min(4.0, (-spin_delta) * 0.015)) | |
| matchup_score += spin_component | |
| if spin_component >= 1.0: | |
| notes.append("Spin down") | |
| if extension_delta is not None: | |
| extension_component = max(-1.5, min(3.0, (-extension_delta) * 2.0)) | |
| matchup_score += extension_component | |
| if extension_component >= 1.0: | |
| notes.append("Extension down") | |
| if pitch_count is not None: | |
| pitch_count_component = max(0.0, min(4.0, (pitch_count - 70.0) * 0.06)) | |
| matchup_score += pitch_count_component | |
| if pitch_count_component >= 1.0: | |
| notes.append("Elevated pitch count") | |
| if times_through_order is not None: | |
| tto_component = max(0.0, min(3.0, (times_through_order - 2.0) * 1.5)) | |
| matchup_score += tto_component | |
| if tto_component >= 1.0: | |
| notes.append("Times-through-order edge") | |
| # ---------------------------- | |
| # Zone matchup | |
| # ---------------------------- | |
| zone_hr_boost = _safe_float(zone_matchup_adj.get("hr_zone_boost")) | |
| zone_hit_boost = _safe_float(zone_matchup_adj.get("hit_zone_boost")) | |
| zone_tb2p_boost = _safe_float(zone_matchup_adj.get("tb2p_zone_boost")) | |
| if zone_hr_boost is not None: | |
| hr_zone_component = max(-3.0, min(8.0, zone_hr_boost * 35.0)) | |
| matchup_score += hr_zone_component | |
| if hr_zone_component >= 1.5: | |
| notes.append("HR zone fit") | |
| if zone_hit_boost is not None: | |
| hit_zone_component = max(-2.0, min(6.0, zone_hit_boost * 20.0)) | |
| matchup_score += hit_zone_component | |
| if hit_zone_component >= 1.0: | |
| notes.append("Hit zone fit") | |
| if zone_tb2p_boost is not None: | |
| tb_zone_component = max(-2.0, min(6.0, zone_tb2p_boost * 20.0)) | |
| matchup_score += tb_zone_component | |
| if tb_zone_component >= 1.0: | |
| notes.append("Power zone fit") | |
| # ---------------------------- | |
| # Environment | |
| # ---------------------------- | |
| temperature_f = _safe_float(temperature_f) | |
| wind_speed_mph = _safe_float(wind_speed_mph) | |
| if temperature_f is not None: | |
| temp_component = max(-2.0, min(3.0, (temperature_f - 72.0) * 0.08)) | |
| matchup_score += temp_component | |
| if temp_component >= 1.0: | |
| notes.append("Warm weather") | |
| if wind_speed_mph is not None: | |
| wind_component = max(0.0, min(2.5, wind_speed_mph * 0.08)) | |
| matchup_score += wind_component | |
| if wind_component >= 1.0: | |
| notes.append("Wind support") | |
| if venue_name: | |
| venue_text = str(venue_name).strip() | |
| if venue_text: | |
| notes.append(f"Venue: {venue_text}") | |
| if trust_live_score is not None and trust_live_score < 0.25: | |
| notes.append("Low live-state confidence") | |
| matchup_score = max(0.0, min(100.0, matchup_score)) | |
| if matchup_score >= 70: | |
| matchup_rating = "Great" | |
| elif matchup_score >= 58: | |
| matchup_rating = "Good" | |
| elif matchup_score >= 45: | |
| matchup_rating = "Neutral" | |
| else: | |
| matchup_rating = "Poor" | |
| return { | |
| "matchup_score": matchup_score, | |
| "matchup_rating": matchup_rating, | |
| "matchup_notes": notes[:5], | |
| } | |
| def compute_zone_matchup_adjustment( | |
| batter_zone_row: dict, | |
| pitcher_zone_row: dict, | |
| ) -> dict[str, float]: | |
| adjustment = { | |
| "hr_zone_boost": 0.0, | |
| "hit_zone_boost": 0.0, | |
| "tb2p_zone_boost": 0.0, | |
| } | |
| pitch_families = ["fastball", "breaking", "offspeed"] | |
| zones = ["heart", "shadow", "chase", "waste"] | |
| total_weight = 0.0 | |
| for family in pitch_families: | |
| family_usage_rate = pitcher_zone_row.get(f"{family}_usage_rate") | |
| if family_usage_rate is None: | |
| continue | |
| try: | |
| family_usage_rate = float(family_usage_rate) | |
| except Exception: | |
| continue | |
| if family_usage_rate <= 0: | |
| continue | |
| for zone in zones: | |
| zone_cond_rate = pitcher_zone_row.get(f"{family}_{zone}_cond_rate") | |
| if zone_cond_rate is None: | |
| continue | |
| try: | |
| zone_cond_rate = float(zone_cond_rate) | |
| except Exception: | |
| continue | |
| if zone_cond_rate <= 0: | |
| continue | |
| weight = family_usage_rate * zone_cond_rate | |
| if weight <= 0: | |
| continue | |
| hr_prob = batter_zone_row.get(f"hr_prob_{family}_{zone}") | |
| hit_prob = batter_zone_row.get(f"hit_prob_{family}_{zone}") | |
| tb2p_prob = batter_zone_row.get(f"tb2p_prob_{family}_{zone}") | |
| if hr_prob is not None: | |
| try: | |
| adjustment["hr_zone_boost"] += weight * float(hr_prob) | |
| except Exception: | |
| pass | |
| if hit_prob is not None: | |
| try: | |
| adjustment["hit_zone_boost"] += weight * float(hit_prob) | |
| except Exception: | |
| pass | |
| if tb2p_prob is not None: | |
| try: | |
| adjustment["tb2p_zone_boost"] += weight * float(tb2p_prob) | |
| except Exception: | |
| pass | |
| total_weight += weight | |
| if total_weight > 0: | |
| adjustment["hr_zone_boost"] /= total_weight | |
| adjustment["hit_zone_boost"] /= total_weight | |
| adjustment["tb2p_zone_boost"] /= total_weight | |
| return adjustment | |
| def compute_family_zone_matchup_adjustment( | |
| batter_family_zone_row: dict, | |
| pitcher_family_zone_row: dict, | |
| ) -> dict[str, float]: | |
| adjustment = { | |
| "family_zone_hr_boost": 0.0, | |
| "family_zone_hit_boost": 0.0, | |
| "family_zone_tb2p_boost": 0.0, | |
| "family_zone_whiff_risk": 0.0, | |
| } | |
| pitch_families = ["fastball", "breaking", "offspeed"] | |
| zones = ["heart", "shadow", "chase", "waste"] | |
| total_weight = 0.0 | |
| for family in pitch_families: | |
| for zone in zones: | |
| usage_overall = _safe_float( | |
| pitcher_family_zone_row.get(f"usage_rate_overall_{family}_{zone}") | |
| ) | |
| if usage_overall is None or usage_overall <= 0: | |
| continue | |
| batter_hr_rate = _safe_float( | |
| batter_family_zone_row.get(f"hr_rate_{family}_{zone}") | |
| ) | |
| batter_hit_rate = _safe_float( | |
| batter_family_zone_row.get(f"hit_rate_{family}_{zone}") | |
| ) | |
| batter_damage_rate = _safe_float( | |
| batter_family_zone_row.get(f"damage_rate_{family}_{zone}") | |
| ) | |
| batter_whiff_rate = _safe_float( | |
| batter_family_zone_row.get(f"whiff_rate_{family}_{zone}") | |
| ) | |
| batter_xwoba = _safe_float( | |
| batter_family_zone_row.get(f"xwoba_{family}_{zone}") | |
| ) | |
| pitcher_hr_allowed = _safe_float( | |
| pitcher_family_zone_row.get(f"hr_allowed_rate_{family}_{zone}") | |
| ) | |
| pitcher_hit_allowed = _safe_float( | |
| pitcher_family_zone_row.get(f"hit_allowed_rate_{family}_{zone}") | |
| ) | |
| pitcher_damage_allowed = _safe_float( | |
| pitcher_family_zone_row.get(f"damage_allowed_rate_{family}_{zone}") | |
| ) | |
| pitcher_whiff_rate = _safe_float( | |
| pitcher_family_zone_row.get(f"whiff_rate_{family}_{zone}") | |
| ) | |
| pitcher_xwoba_allowed = _safe_float( | |
| pitcher_family_zone_row.get(f"xwoba_allowed_{family}_{zone}") | |
| ) | |
| hr_signal_parts = [x for x in [batter_hr_rate, pitcher_hr_allowed] if x is not None] | |
| hit_signal_parts = [x for x in [batter_hit_rate, pitcher_hit_allowed] if x is not None] | |
| damage_signal_parts = [x for x in [batter_damage_rate, pitcher_damage_allowed] if x is not None] | |
| whiff_signal_parts = [x for x in [batter_whiff_rate, pitcher_whiff_rate] if x is not None] | |
| xwoba_signal_parts = [x for x in [batter_xwoba, pitcher_xwoba_allowed] if x is not None] | |
| if not hr_signal_parts and not hit_signal_parts and not damage_signal_parts: | |
| continue | |
| hr_signal = sum(hr_signal_parts) / len(hr_signal_parts) if hr_signal_parts else 0.0 | |
| hit_signal = sum(hit_signal_parts) / len(hit_signal_parts) if hit_signal_parts else 0.0 | |
| damage_signal = sum(damage_signal_parts) / len(damage_signal_parts) if damage_signal_parts else 0.0 | |
| whiff_signal = sum(whiff_signal_parts) / len(whiff_signal_parts) if whiff_signal_parts else 0.0 | |
| xwoba_signal = sum(xwoba_signal_parts) / len(xwoba_signal_parts) if xwoba_signal_parts else 0.0 | |
| tb_signal = damage_signal | |
| if xwoba_signal > 0: | |
| tb_signal += max(-0.05, min(0.10, (xwoba_signal - 0.320) * 0.50)) | |
| adjustment["family_zone_hr_boost"] += usage_overall * hr_signal | |
| adjustment["family_zone_hit_boost"] += usage_overall * hit_signal | |
| adjustment["family_zone_tb2p_boost"] += usage_overall * tb_signal | |
| adjustment["family_zone_whiff_risk"] += usage_overall * whiff_signal | |
| total_weight += usage_overall | |
| if total_weight > 0: | |
| adjustment["family_zone_hr_boost"] /= total_weight | |
| adjustment["family_zone_hit_boost"] /= total_weight | |
| adjustment["family_zone_tb2p_boost"] /= total_weight | |
| adjustment["family_zone_whiff_risk"] /= total_weight | |
| adjustment["sample_size"] = total_weight | |
| return adjustment |