Create upcoming_edge_stub.py

models/upcoming_edge_stub.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pandas as pd
+
+
+def _safe_float(value: Any, default: float = 0.0) -> float:
+    try:
+        if value is None:
+            return default
+        text = str(value).strip().lower()
+        if text in {"", "nan", "none"}:
+            return default
+        return float(value)
+    except Exception:
+        return default
+
+
+def probability_to_american(prob: float) -> int | None:
+    if prob <= 0 or prob >= 1:
+        return None
+
+    if prob >= 0.5:
+        return int(round(-(prob / (1 - prob)) * 100))
+    return int(round(((1 - prob) / prob) * 100))
+
+
+def implied_prob_from_american(odds: int) -> float:
+    if odds > 0:
+        return 100 / (odds + 100)
+    return abs(odds) / (abs(odds) + 100)
+
+
+def _stub_probs_for_batter(
+    batter_name: str,
+    statcast_df: pd.DataFrame,
+    game_row: dict[str, Any],
+) -> dict[str, float]:
+    hit_prob = 0.24
+    hr_prob = 0.045
+    tb2p_prob = 0.17
+
+    balls = _safe_float(game_row.get("balls"), 0.0)
+    strikes = _safe_float(game_row.get("strikes"), 0.0)
+    outs = _safe_float(game_row.get("outs"), 0.0)
+
+    runner_on_2b = bool(game_row.get("runner_on_2b", False))
+    runner_on_3b = bool(game_row.get("runner_on_3b", False))
+
+    if batter_name and not statcast_df.empty and "player_name" in statcast_df.columns:
+        batter_rows = statcast_df[statcast_df["player_name"].astype(str) == batter_name].copy()
+
+        if not batter_rows.empty:
+            if "launch_speed" in batter_rows.columns:
+                avg_ev = pd.to_numeric(batter_rows["launch_speed"], errors="coerce").dropna()
+                if not avg_ev.empty:
+                    hit_prob += min(0.05, max(-0.03, (avg_ev.mean() - 88.0) * 0.002))
+                    hr_prob += min(0.03, max(-0.015, (avg_ev.mean() - 90.0) * 0.0015))
+                    tb2p_prob += min(0.04, max(-0.02, (avg_ev.mean() - 88.0) * 0.0015))
+
+    if balls >= 2:
+        hit_prob += 0.015
+        tb2p_prob += 0.01
+    if strikes >= 2:
+        hit_prob -= 0.015
+        hr_prob -= 0.005
+
+    if runner_on_2b or runner_on_3b:
+        hit_prob += 0.01
+    if outs >= 2:
+        hit_prob -= 0.005
+        tb2p_prob -= 0.005
+
+    hit_prob = max(0.05, min(0.50, hit_prob))
+    hr_prob = max(0.005, min(0.20, hr_prob))
+    tb2p_prob = max(0.03, min(0.40, tb2p_prob))
+
+    return {
+        "hit_prob": hit_prob,
+        "hr_prob": hr_prob,
+        "tb2p_prob": tb2p_prob,
+    }
+
+
+def _market_row(label: str, batter_name: str, probs: dict[str, float]) -> dict[str, Any]:
+    fair_hit = probability_to_american(probs["hit_prob"])
+    fair_hr = probability_to_american(probs["hr_prob"])
+    fair_tb2p = probability_to_american(probs["tb2p_prob"])
+
+    book_hit = +135
+    book_hr = +425
+    book_tb2p = +165
+
+    return {
+        "slot": label,
+        "batter_name": batter_name,
+        "fair_hit_odds": fair_hit,
+        "fair_hr_odds": fair_hr,
+        "fair_tb2p_odds": fair_tb2p,
+        "book_hit_odds": book_hit,
+        "book_hr_odds": book_hr,
+        "book_tb2p_odds": book_tb2p,
+        "hit_edge": probs["hit_prob"] - implied_prob_from_american(book_hit),
+        "hr_edge": probs["hr_prob"] - implied_prob_from_american(book_hr),
+        "tb2p_edge": probs["tb2p_prob"] - implied_prob_from_american(book_tb2p),
+    }
+
+
+def compute_upcoming_edge_rows(
+    game_row: dict[str, Any],
+    statcast_df: pd.DataFrame,
+) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+
+    slots = [
+        ("On Deck", str(game_row.get("on_deck_name", "") or "").strip()),
+        ("In Hole", str(game_row.get("in_hole_name", "") or "").strip()),
+        ("3 Away", str(game_row.get("three_away_name", "") or "").strip()),
+    ]
+
+    for label, batter_name in slots:
+        if not batter_name:
+            continue
+        probs = _stub_probs_for_batter(batter_name, statcast_df=statcast_df, game_row=game_row)
+        rows.append(_market_row(label, batter_name, probs))
+
+    return rows