"""
visualization/props_page.py
Props page for upcoming HR props.
This module now provides a decomposed layout scaffold for the upcoming Props
redesign: hero summary, featured props, by-game explorer, and a legacy flat
table for continuity while the premium UI is built out.
"""
from __future__ import annotations
import datetime
import threading
from typing import Any
import pandas as pd
import streamlit as st
from analytics.props_mapper import map_props_to_models
from analytics.props_view_model import (
build_best_on_slate_df,
build_best_on_slate_summary,
build_hr_props_view_model,
select_best_lines_per_prop,
)
from data.shared_baseline import load_or_build_shared_baseline_bundle_complete_for_request
from data.live_prop_odds import fetch_all_upcoming_hr_props
from database.db import (
ensure_upcoming_hr_props_table,
get_connection,
insert_upcoming_hr_props,
)
from utils.helpers import utc_now_iso
from data.mlb_starters import (
_canonical_team,
build_oddsapi_starter_fallback_map,
lookup_pitchers_for_game,
merge_probable_starters_with_odds_fallback,
)
_PROPS_ASYNC_LOCK = threading.Lock()
_PROPS_ASYNC_KEYS: set[str] = set()
def _queue_props_async_refresh(key: str, fn) -> bool:
with _PROPS_ASYNC_LOCK:
if key in _PROPS_ASYNC_KEYS:
return False
_PROPS_ASYNC_KEYS.add(key)
def _run() -> None:
try:
fn()
finally:
with _PROPS_ASYNC_LOCK:
_PROPS_ASYNC_KEYS.discard(key)
threading.Thread(target=_run, daemon=True).start()
return True
def _render_props_ui_styles() -> None:
st.markdown(
"""
""",
unsafe_allow_html=True,
)
def _format_pct(val: float | None) -> str:
if val is None:
return "-"
return f"{val * 100:.1f}%"
def _format_odds(val: int | float | None) -> str:
if val is None:
return "-"
v = int(val)
return f"+{v}" if v > 0 else str(v)
def _format_edge(val: float | None) -> str:
if val is None:
return "-"
return f"{val * 100:+.1f}%"
def _format_ev(val: float | None) -> str:
if val is None:
return "-"
return f"{val:+.2f}u"
def _format_confidence(val: float | None) -> str:
if val is None:
return "-"
try:
return f"{float(val):.0f}"
except Exception:
return "-"
def _confidence_summary_label(value: Any) -> str:
text = str(value or "").strip()
return text if text else "-"
def _build_confidence_metric_html(row: pd.Series | dict[str, Any]) -> str:
score = _format_confidence(row.get("confidence_score"))
summary = _confidence_summary_label(row.get("confidence_summary_label"))
if summary == "-":
summary = " "
return (
f"{score}
"
f"{summary}
"
)
def _render_confidence_breakdown(details_row: dict[str, Any]) -> None:
raw_score = details_row.get("confidence_score_raw")
display_score = details_row.get("confidence_score_display", details_row.get("confidence_score"))
confidence_source = str(details_row.get("confidence_source") or "").strip() or "-"
final_bucket = str(details_row.get("confidence_bucket") or details_row.get("confidence_bucket_display") or "-").strip()
summary_label = str(details_row.get("confidence_summary_label") or "").strip()
bonuses = details_row.get("confidence_component_bonuses") or []
penalties = details_row.get("confidence_component_penalties") or []
if (
raw_score is None
and display_score is None
and confidence_source == "-"
and not bonuses
and not penalties
):
return
st.caption("Confidence Breakdown")
metric_cols = st.columns(4)
metric_cols[0].metric("Raw/Base", _format_confidence(raw_score))
metric_cols[1].metric("Display", _format_confidence(display_score))
metric_cols[2].metric("Source", confidence_source.replace("_", " ").title())
metric_cols[3].metric("Bucket", final_bucket.title() if final_bucket else "-")
if bonuses:
st.write("Bonuses")
for bonus in bonuses:
st.write(f"- +{float(bonus.get('value') or 0.0):.0f} {str(bonus.get('label') or '-').strip()}")
if penalties:
st.write("Penalties")
for penalty in penalties:
value = float(penalty.get("value") or 0.0)
prefix = f"-{value:.0f}" if value > 0 else "0"
st.write(f"- {prefix} {str(penalty.get('label') or '-').strip()}")
why_lines: list[str] = []
if summary_label:
why_lines.append(f"Primary driver: {summary_label}")
for reason in details_row.get("confidence_reasons") or []:
if str(reason).strip():
why_lines.append(str(reason).strip())
if why_lines:
st.write("Why It Landed Here")
for line in why_lines:
st.write(f"- {line}")
def _market_label(value: Any) -> str:
text = str(value or "").strip().lower()
labels = {
"hr": "1+ Home Runs",
"k": "Pitcher Strikeouts",
"no_hr": "No Home Run",
}
return labels.get(text, text.upper() if text else "-")
def _metric_tone_class(metric: str, value: float | None) -> str:
if value is None:
return ""
try:
numeric = float(value)
except Exception:
return ""
if metric == "ev":
if numeric >= 0.05:
return "good"
if numeric >= -0.05:
return "neutral"
return "bad"
if metric == "edge":
if numeric >= 0.01:
return "good"
if numeric >= -0.01:
return "neutral"
return "bad"
return ""
def _render_verdict_badge(verdict: Any) -> str:
text = str(verdict or "tracked").strip().lower()
label = text.upper()
return f'{label}
'
def _render_props_legend() -> None:
st.markdown(
"""
Legend
Odds: the sportsbook price.
Implied: sportsbook break-even probability from those odds.
Pregame HR%: our modeled fair probability for the market.
Edge: model probability minus implied probability.
EV: expected units won or lost per 1u staked.
Confidence: 1-100 trust score based on sample size and signal quality.
Bet: positive value with enough confidence to act.
Watch / Pass: monitor or avoid; value is weak or negative.
""",
unsafe_allow_html=True,
)
def _style_metric_dataframe(df: pd.DataFrame) -> pd.io.formats.style.Styler:
def style_val(value: Any, metric: str) -> str:
if value is None:
return ""
text = str(value).strip()
if text in {"", "-", "nan", "None"}:
return ""
try:
numeric_text = text.replace("%", "").replace("u", "").replace("+", "")
numeric = float(numeric_text)
except Exception:
return ""
tone = _metric_tone_class(metric, numeric / 100.0 if metric == "edge" else numeric)
color_map = {
"good": "#6ef08d",
"neutral": "#f3cd65",
"bad": "#ff7f7f",
}
color = color_map.get(tone)
return f"color: {color}; font-weight: 700;" if color else ""
styler = df.style
if "edge" in df.columns:
styler = styler.map(lambda v: style_val(v, "edge"), subset=["edge"])
if "bet_ev" in df.columns:
styler = styler.map(lambda v: style_val(v, "ev"), subset=["bet_ev"])
if "EV" in df.columns:
styler = styler.map(lambda v: style_val(v, "ev"), subset=["EV"])
if "Edge" in df.columns:
styler = styler.map(lambda v: style_val(v, "edge"), subset=["Edge"])
return styler
def _build_matchup(row: pd.Series | dict[str, Any]) -> str:
away = str(row.get("away_team") or "")
home = str(row.get("home_team") or "")
if away and home:
return f"{away} @ {home}"
return away or home or "-"
def _build_game_time(row: pd.Series | dict[str, Any]) -> str:
ct = str(row.get("commence_time") or "")
if not ct:
return "-"
try:
dt = datetime.datetime.fromisoformat(ct.replace("Z", "+00:00"))
return dt.strftime("%m/%d %I:%M%p UTC")
except Exception:
return ct[:16]
def _display_market_line_label(row: pd.Series | dict[str, Any]) -> str:
display_label = str(row.get("display_label") or "").strip()
market_family = str(row.get("market_family") or row.get("market") or "").strip().lower()
if market_family == "k":
selection_side = str(row.get("selection_side") or "").strip().lower()
line = row.get("line")
if line is not None and str(line).strip() not in {"", "nan", "None"}:
try:
line_text = f"{float(line):.1f}"
except Exception:
line_text = str(line).strip()
if selection_side in {"over", "under"}:
return f"{selection_side.title()} {line_text} Ks"
if display_label and display_label.casefold() != "pitcher strikeouts":
return display_label
return "Pitcher Strikeouts"
if display_label:
return display_label
return "1+ HR" if market_family == "hr" else _market_label(market_family)
def _normalize_team_key(value: Any) -> str:
return " ".join(str(value or "").strip().lower().split())
def _extract_slate_teams(filtered_raw: pd.DataFrame) -> tuple[str, ...]:
if filtered_raw is None or filtered_raw.empty:
return tuple()
teams: set[str] = set()
for col in ("away_team", "home_team"):
if col in filtered_raw.columns:
teams.update(
{
str(team).strip()
for team in filtered_raw[col].dropna().astype(str).tolist()
if str(team).strip()
}
)
return tuple(sorted(teams))
def _extract_slate_matchups(filtered_raw: pd.DataFrame) -> tuple[tuple[str, str], ...]:
if filtered_raw is None or filtered_raw.empty:
return tuple()
matchups: set[tuple[str, str]] = set()
cols_present = {"away_team", "home_team"}.issubset(filtered_raw.columns)
if not cols_present:
return tuple()
for _, row in filtered_raw[["away_team", "home_team"]].dropna(how="any").iterrows():
away_team = str(row.get("away_team") or "").strip()
home_team = str(row.get("home_team") or "").strip()
if away_team and home_team:
matchups.add((away_team, home_team))
return tuple(sorted(matchups))
def _extract_oddsapi_pitcher_names(raw: pd.DataFrame | None) -> tuple[str, ...]:
if raw is None or raw.empty:
return tuple()
market_series = raw.get("market_family", raw.get("market", pd.Series(dtype="object", index=raw.index)))
scope_series = raw.get("selection_scope", pd.Series(dtype="object", index=raw.index))
k_rows = raw[
market_series.fillna("").astype(str).str.strip().str.lower().eq("k")
& scope_series.fillna("").astype(str).str.strip().str.lower().eq("pitcher")
].copy()
if k_rows.empty:
return tuple()
names = {
str(name).strip()
for name in k_rows.get("player_name_raw", pd.Series(dtype="object")).dropna().astype(str).tolist()
if str(name).strip()
}
return tuple(sorted(names))
def _extract_props_hitter_names(raw: pd.DataFrame | None) -> tuple[str, ...]:
if raw is None or raw.empty or "player_name" not in raw.columns:
return tuple()
market_series = raw.get("market_family", raw.get("market", pd.Series(dtype="object", index=raw.index)))
hitter_rows = raw[~market_series.fillna("").astype(str).str.strip().str.lower().eq("k")].copy()
if hitter_rows.empty:
return tuple()
names = {
str(name).strip()
for name in hitter_rows["player_name"].dropna().astype(str).tolist()
if str(name).strip()
}
return tuple(sorted(names))
def _extract_starter_pitcher_names(starters_map: dict | None) -> tuple[str, ...]:
if not starters_map:
return tuple()
names: set[str] = set()
for payload in starters_map.values():
if not isinstance(payload, dict):
continue
for key in ("away_pitcher", "home_pitcher"):
pitcher_name = str(payload.get(key) or "").strip()
if pitcher_name:
names.add(pitcher_name)
return tuple(sorted(names))
def _supported_props_markets(raw: pd.DataFrame | None) -> tuple[str, ...]:
if raw is None or raw.empty or "market" not in raw.columns:
return tuple()
return tuple(
market
for market in sorted(raw["market"].dropna().astype(str).str.lower().unique().tolist())
if market in {"hr", "k"}
)
def _filter_probable_starters_to_slate(
probable_starters: dict | None,
slate_teams: tuple[str, ...],
) -> dict:
if not probable_starters or not slate_teams:
return {}
team_scope = {_canonical_team(team) for team in slate_teams if str(team).strip()}
out: dict = {}
for key, payload in probable_starters.items():
if not isinstance(key, tuple) or len(key) != 2:
continue
away_norm = _canonical_team(key[0])
home_norm = _canonical_team(key[1])
if away_norm in team_scope and home_norm in team_scope:
out[key] = payload
return out
@st.cache_data(ttl=60 * 10, show_spinner=False)
def _load_props_prepared_bundle(
raw: pd.DataFrame,
probable_starters: dict | None,
) -> dict[str, Any]:
slate_teams = _extract_slate_teams(raw)
supported_markets = _supported_props_markets(raw)
primary_starters = _filter_probable_starters_to_slate(probable_starters, slate_teams)
hitter_names = _extract_props_hitter_names(raw)
pitcher_names = tuple(
sorted(
set(_extract_oddsapi_pitcher_names(raw))
| set(_extract_starter_pitcher_names(primary_starters))
)
)
bundle = load_or_build_shared_baseline_bundle_complete_for_request(
batter_names=hitter_names,
pitcher_names=pitcher_names,
max_age_seconds=60 * 60,
persist_runtime_refresh=True,
)
pitcher_statcast_df = bundle.get("blended_pitcher_df", pd.DataFrame())
starter_bundle = _load_props_starter_bundle(
raw=raw,
probable_starters=primary_starters,
pitcher_statcast_df=pitcher_statcast_df,
)
prepared_bundle = {
"slate_team_scope": slate_teams,
"supported_markets": supported_markets,
"requested_hitter_names": hitter_names,
"requested_pitcher_names": pitcher_names,
"blended_batter_df": bundle.get("blended_batter_df", pd.DataFrame()),
"blended_pitcher_df": pitcher_statcast_df,
"batter_baseline_meta": bundle.get("batter_baseline_meta", pd.DataFrame()),
"pitcher_baseline_meta": bundle.get("pitcher_baseline_meta", pd.DataFrame()),
"snapshot_status": bundle.get("snapshot_status", pd.DataFrame()),
"snapshot_source_status": bundle.get("snapshot_source_status"),
"runtime_fallback_used": bundle.get("runtime_fallback_used"),
"requested_hitter_count": bundle.get("requested_hitter_count"),
"requested_pitcher_count": bundle.get("requested_pitcher_count"),
"resolved_hitter_count": bundle.get("resolved_hitter_count"),
"resolved_pitcher_count": bundle.get("resolved_pitcher_count"),
"missing_hitter_names": bundle.get("missing_hitter_names", []),
"missing_pitcher_names": bundle.get("missing_pitcher_names", []),
"snapshot_coverage_mode": bundle.get("snapshot_coverage_mode"),
"background_refresh_queued": bundle.get("background_refresh_queued"),
"request_patch_used": bundle.get("request_patch_used"),
"starter_bundle": starter_bundle,
"starter_debug": {
"starter_cache_source": starter_bundle.get("starter_cache_source") or "unresolved",
"oddsapi_fallback_used_matchup_count": int(starter_bundle.get("oddsapi_fallback_used_matchup_count") or 0),
},
}
prepared_bundle["signature"] = (
int(len(raw)),
int(raw["event_id"].nunique()) if "event_id" in raw.columns else 0,
supported_markets,
int(prepared_bundle.get("requested_hitter_count") or len(hitter_names)),
int(prepared_bundle.get("requested_pitcher_count") or len(pitcher_names)),
int(prepared_bundle.get("resolved_hitter_count") or 0),
int(prepared_bundle.get("resolved_pitcher_count") or 0),
str(prepared_bundle.get("snapshot_source_status") or ""),
str(starter_bundle.get("starter_cache_source") or ""),
int(starter_bundle.get("oddsapi_fallback_used_matchup_count") or 0),
)
return prepared_bundle
def _build_props_baseline_debug(
*,
market_type: str,
slate_team_scope: tuple[str, ...],
prepared_bundle: dict[str, Any],
) -> dict[str, Any]:
return {
"market_type": market_type,
"slate_team_scope": list(slate_team_scope),
"requested_hitter_count": int(prepared_bundle.get("requested_hitter_count", len(prepared_bundle.get("requested_hitter_names") or ()))),
"requested_pitcher_count": int(prepared_bundle.get("requested_pitcher_count", len(prepared_bundle.get("requested_pitcher_names") or ()))),
"resolved_hitter_count": int(prepared_bundle.get("resolved_hitter_count", 0)),
"resolved_pitcher_count": int(prepared_bundle.get("resolved_pitcher_count", 0)),
"missing_hitter_names": list(prepared_bundle.get("missing_hitter_names", [])),
"missing_pitcher_names": list(prepared_bundle.get("missing_pitcher_names", [])),
"snapshot_coverage_mode": str(prepared_bundle.get("snapshot_coverage_mode") or "unknown"),
"runtime_fallback_used": bool(prepared_bundle.get("runtime_fallback_used")),
"background_refresh_queued": bool(prepared_bundle.get("background_refresh_queued")),
"request_patch_used": bool(prepared_bundle.get("request_patch_used")),
"baseline_source": str(prepared_bundle.get("snapshot_source_status") or "unknown"),
}
def _build_market_modeling_payload(
*,
filtered_raw: pd.DataFrame,
market_type: str,
prepared_bundle: dict[str, Any],
capture_debug: bool = False,
) -> dict[str, Any]:
slate_team_scope = _extract_slate_teams(filtered_raw)
scoped_probable_starters = _filter_probable_starters_to_slate(
(prepared_bundle.get("starter_bundle") or {}).get("merged_starters"),
slate_team_scope,
)
baseline_debug = _build_props_baseline_debug(
market_type=market_type,
slate_team_scope=slate_team_scope,
prepared_bundle=prepared_bundle,
)
if capture_debug:
st.session_state["props_baseline_debug"] = baseline_debug
statcast_df = prepared_bundle.get("blended_batter_df", pd.DataFrame())
pitcher_statcast_df = prepared_bundle.get("blended_pitcher_df", pd.DataFrame())
if (statcast_df is None or statcast_df.empty) and market_type in {"hr", "hit", "tb"}:
statcast_df = pd.DataFrame()
if pitcher_statcast_df is None:
pitcher_statcast_df = pd.DataFrame()
mapped = _map_market_rows(
market_type=market_type,
filtered_raw=filtered_raw,
statcast_df=statcast_df,
pitcher_statcast_df=pitcher_statcast_df,
probable_starters=scoped_probable_starters,
)
if capture_debug:
if market_type == "hr":
st.session_state["props_hr_health_debug"] = _build_hr_health_debug(
mapped,
extra_context={
"requested_hitter_count": baseline_debug.get("requested_hitter_count"),
"resolved_hitter_count": baseline_debug.get("resolved_hitter_count"),
"requested_pitcher_count": baseline_debug.get("requested_pitcher_count"),
"resolved_pitcher_count": baseline_debug.get("resolved_pitcher_count"),
},
)
else:
st.session_state.pop("props_hr_health_debug", None)
st.session_state["props_shared_component_debug"] = _build_shared_component_debug(
mapped,
market_type=market_type,
)
return {
"baseline_debug": baseline_debug,
"statcast_df": statcast_df,
"pitcher_statcast_df": pitcher_statcast_df,
"mapped": mapped,
}
def _build_props_market_debug_payload(
*,
market_type: str,
payload: dict[str, Any],
) -> dict[str, Any]:
baseline_debug = dict(payload.get("baseline_debug") or {})
mapped = payload.get("mapped", pd.DataFrame())
hr_health = None
if market_type == "hr":
hr_health = _build_hr_health_debug(
mapped,
extra_context={
"requested_hitter_count": baseline_debug.get("requested_hitter_count"),
"resolved_hitter_count": baseline_debug.get("resolved_hitter_count"),
"requested_pitcher_count": baseline_debug.get("requested_pitcher_count"),
"resolved_pitcher_count": baseline_debug.get("resolved_pitcher_count"),
},
)
return {
"market_type": market_type,
"baseline_debug": baseline_debug,
"hr_health_debug": hr_health,
"shared_component_debug": _build_shared_component_debug(mapped, market_type=market_type),
"mapped": mapped,
}
def _build_market_payload_from_prepared_bundle(
*,
raw: pd.DataFrame,
market_type: str,
prepared_bundle: dict[str, Any],
capture_debug: bool = False,
) -> dict[str, Any] | None:
if raw is None or raw.empty:
return None
filtered_raw = raw[raw["market"].astype(str).str.lower() == str(market_type or "").strip().lower()].copy()
if filtered_raw.empty:
return None
return _build_market_modeling_payload(
filtered_raw=filtered_raw,
market_type=market_type,
prepared_bundle=prepared_bundle,
capture_debug=capture_debug,
)
def _ensure_props_market_payloads(
*,
raw: pd.DataFrame,
prepared_bundle: dict[str, Any],
existing_payloads: dict[str, dict[str, Any]] | None = None,
markets: tuple[str, ...] | list[str] | None = None,
capture_debug: bool = False,
) -> dict[str, dict[str, Any]]:
payloads = dict(existing_payloads or {})
target_markets = tuple(
str(market).strip().lower()
for market in (markets or prepared_bundle.get("supported_markets") or ())
if str(market).strip()
)
for market in target_markets:
if market in payloads:
continue
payload = _build_market_payload_from_prepared_bundle(
raw=raw,
market_type=market,
prepared_bundle=prepared_bundle,
capture_debug=capture_debug,
)
if payload is not None:
payloads[market] = payload
return payloads
def _hydrate_props_debug_state(
*,
market_type: str,
payload: dict[str, Any],
) -> None:
st.session_state["props_baseline_debug"] = dict(payload.get("baseline_debug") or {"market_type": market_type})
if market_type == "hr":
st.session_state["props_hr_health_debug"] = {}
else:
st.session_state.pop("props_hr_health_debug", None)
st.session_state["props_shared_component_debug"] = {}
def _build_best_on_slate_source(
*,
modeled_market_bundle: dict[str, dict[str, Any]],
active_market_type: str,
active_mapped: pd.DataFrame,
) -> pd.DataFrame:
if not modeled_market_bundle:
return pd.DataFrame()
market_frames: list[pd.DataFrame] = []
for market, payload in modeled_market_bundle.items():
if market == active_market_type:
if active_mapped is not None and not active_mapped.empty:
market_frames.append(active_mapped.copy())
continue
market_mapped = payload.get("mapped", pd.DataFrame())
if market_mapped is not None and not market_mapped.empty:
market_frames.append(market_mapped.copy())
if not market_frames:
return pd.DataFrame()
return pd.concat(market_frames, ignore_index=True, sort=False)
def _load_raw_props(raw_props: pd.DataFrame | None) -> pd.DataFrame:
if raw_props is not None:
return raw_props
with st.spinner("Loading props markets..."):
return fetch_all_upcoming_hr_props(sportsbooks=DEFAULT_PROP_BOOKS)
def _load_props_starter_bundle(
raw: pd.DataFrame,
probable_starters: dict | None,
pitcher_statcast_df: pd.DataFrame | None = None,
) -> dict[str, Any]:
slate_matchups = _extract_slate_matchups(raw)
primary_starters = dict(probable_starters or {})
fallback_map = build_oddsapi_starter_fallback_map(
props_feed=raw,
primary_starters=primary_starters,
pitcher_statcast_df=pitcher_statcast_df,
)
merged = merge_probable_starters_with_odds_fallback(primary_starters, fallback_map)
cache_sources = {
str(payload.get("starter_cache_source") or "").strip()
for payload in merged.values()
if str(payload.get("starter_cache_source") or "").strip()
}
if "statsapi_plus_oddsapi_fallback" in cache_sources:
aggregate_source = "statsapi_plus_oddsapi_fallback"
elif any(source.startswith("oddsapi_") for source in cache_sources):
aggregate_source = "oddsapi_pitcher_strikeouts_fallback"
elif "statsapi_probable_pitcher" in cache_sources:
aggregate_source = "statsapi_probable_pitcher"
else:
aggregate_source = "unresolved"
return {
"matchups": slate_matchups,
"primary_starters": primary_starters,
"oddsapi_fallback_starters": fallback_map,
"merged_starters": merged,
"starter_cache_source": aggregate_source,
"oddsapi_fallback_used_matchup_count": sum(
1 for payload in merged.values() if bool(payload.get("fallback_used"))
),
}
def _render_empty_props_state() -> None:
st.warning("No supported props returned from The Odds API.")
with st.expander("Diagnostic details"):
st.write("**Step 1 endpoint:**", "https://api.the-odds-api.com/v4/sports/baseball_mlb/events")
st.write("**Step 2 endpoint:**", "https://api.the-odds-api.com/v4/sports/baseball_mlb/events/{eventId}/odds")
st.write("**Markets:**", "batter_home_runs, pitcher_strikeouts")
st.write("**Books requested:**", DEFAULT_PROP_BOOKS)
st.write(
"**Check the terminal/Streamlit log for lines tagged** `[upcoming_hr_props]` "
"to see HTTP status, remaining credits, and per-game bookmaker structure."
)
st.write(
"Possible causes: (1) no upcoming MLB events in the next 7 days, "
"(2) HTTP error on events list or per-event odds call (see log), "
"(3) books requested not covering this market."
)
def _map_market_rows(
market_type: str,
filtered_raw: pd.DataFrame,
statcast_df: pd.DataFrame,
pitcher_statcast_df: pd.DataFrame | None,
probable_starters: dict | None,
) -> pd.DataFrame:
mapped = map_props_to_models(
filtered_raw,
statcast_df,
pitcher_statcast_df=pitcher_statcast_df,
probable_starters=probable_starters,
)
if market_type == "hr" and not mapped.empty:
st.session_state["props_exec_df"] = mapped
return mapped
def _maybe_log_props(conn, mapped: pd.DataFrame) -> None:
if conn is None or mapped.empty:
return
try:
to_log = mapped.copy()
to_log["fetched_at"] = utc_now_iso()
log_cols = [
"fetched_at", "event_id", "commence_time", "away_team", "home_team",
"sportsbook", "market", "market_variant", "threshold", "display_label",
"is_primary_line", "is_modeled", "selection_scope", "selection_side",
"player_name_raw", "player_name",
"odds_american", "line", "implied_prob", "raw_hr_prob",
"calibrated_hr_prob", "model_hr_prob", "fair_prob", "bet_ev", "confidence_score",
"confidence_bucket", "opportunity_hr_adjustment", "model_hr_prob_source", "edge",
"verdict", "model_voice", "model_voice_primary_reason", "model_voice_caveat", "model_voice_tags", "model_voice_for", "model_voice_against",
]
for col in log_cols:
if col not in to_log.columns:
to_log[col] = None
if "model_voice_tags" in to_log.columns:
to_log["model_voice_tags"] = to_log["model_voice_tags"].apply(
lambda v: "|".join(v) if isinstance(v, list) else v
)
ensure_upcoming_hr_props_table(conn)
insert_upcoming_hr_props(conn, to_log[log_cols])
except Exception:
pass
def _render_filter_controls(mapped: pd.DataFrame, market_type: str) -> tuple[list[str], float, str, str]:
all_books = sorted(mapped["sportsbook"].dropna().unique().tolist()) if "sportsbook" in mapped.columns else []
filter_cols = st.columns([2.2, 1.1, 1.1, 1.2])
with filter_cols[0]:
selected_books = st.multiselect(
"Sportsbook",
options=all_books,
default=all_books,
key="props_books",
)
if market_type == "hr":
with filter_cols[1]:
min_edge = st.slider(
"Min edge",
min_value=-0.50,
max_value=0.50,
value=-0.50,
step=0.01,
format="%.0f%%",
key="props_min_edge",
)
with filter_cols[2]:
sort_option = st.selectbox(
"Sort by",
options=["EV", "Edge", "Confidence", "Pregame Probability"],
index=0,
key="props_sort",
)
else:
min_edge = -0.50
with filter_cols[1]:
st.write("")
with filter_cols[2]:
sort_option = st.selectbox(
"Sort by",
options=["EV", "Edge", "Confidence", "Implied Probability"],
index=0,
key="props_sort",
)
with filter_cols[3]:
view = st.radio("View", ["All Books", "Best Line"], horizontal=False, key="props_view")
return selected_books, min_edge, sort_option, view
def _get_current_filter_state(mapped: pd.DataFrame, market_type: str) -> tuple[list[str], float, str, str]:
all_books = sorted(mapped["sportsbook"].dropna().unique().tolist()) if "sportsbook" in mapped.columns else []
selected_books = st.session_state.get("props_books", all_books)
if not isinstance(selected_books, list) or not selected_books:
selected_books = all_books
selected_books = [book for book in selected_books if book in all_books] or all_books
if market_type == "hr":
min_edge = float(st.session_state.get("props_min_edge", -0.50))
sort_option = str(st.session_state.get("props_sort", "EV"))
else:
min_edge = -0.50
sort_option = str(st.session_state.get("props_sort", "EV"))
view = str(st.session_state.get("props_view", "All Books"))
if view not in {"All Books", "Best Line"}:
view = "All Books"
return selected_books, min_edge, sort_option, view
def _render_market_coverage_note(display: pd.DataFrame, market_type: str) -> None:
if display is None or display.empty or "sportsbook" not in display.columns:
return
available_books = sorted(display["sportsbook"].dropna().astype(str).unique().tolist())
if not available_books:
return
label = _market_label(market_type)
bundle_debug = st.session_state.get("upcoming_props_bundle_debug") or {}
hr_completeness = dict(bundle_debug.get("hr_snapshot_completeness") or {})
hr_snapshot_state = str(bundle_debug.get("hr_snapshot_state") or "").strip().lower()
adapter_status_by_book = dict(bundle_debug.get("adapter_status_by_book") or {})
adapter_retry_after_by_book = dict(bundle_debug.get("adapter_retry_after_by_book") or {})
cache_source = str(bundle_debug.get("cache_source") or "")
if len(available_books) == 1:
st.info(f"{label} odds currently available from {available_books[0]} only on this slate.")
if market_type == "hr" and hr_completeness and not bool(hr_completeness.get("is_complete", True)):
missing_books = ", ".join(hr_completeness.get("missing_books") or [])
refresh_text = "Background refresh is trying to backfill missing HR book coverage."
if missing_books:
refresh_text += f" Still missing: {missing_books}."
dk_status = str(adapter_status_by_book.get("draftkings") or "").strip()
dk_retry_after = str(adapter_retry_after_by_book.get("draftkings") or "").strip()
if dk_status:
refresh_text += f" DraftKings adapter: {dk_status.replace('_', ' ')}."
if dk_retry_after:
refresh_text += f" Next DK retry after: {dk_retry_after}."
if hr_snapshot_state:
refresh_text += f" HR snapshot state: {hr_snapshot_state.replace('_', ' ')}."
if cache_source:
refresh_text += f" Source: {cache_source.replace('_', ' ')}."
st.caption(refresh_text)
return
books_text = ", ".join(available_books)
st.caption(f"{label} coverage on this slate: {books_text}")
if market_type == "hr" and hr_completeness and not bool(hr_completeness.get("is_complete", True)):
missing_books = ", ".join(hr_completeness.get("missing_books") or [])
refresh_text = "HR coverage is still incomplete for this slate."
if missing_books:
refresh_text += f" Missing books: {missing_books}."
dk_status = str(adapter_status_by_book.get("draftkings") or "").strip()
if dk_status:
refresh_text += f" DraftKings adapter: {dk_status.replace('_', ' ')}."
if hr_snapshot_state:
refresh_text += f" HR snapshot state: {hr_snapshot_state.replace('_', ' ')}."
if cache_source:
refresh_text += f" Source: {cache_source.replace('_', ' ')}."
st.caption(refresh_text)
def _prepare_display_frames(
mapped: pd.DataFrame,
market_type: str,
selected_books: list[str],
min_edge: float,
sort_option: str,
view: str,
) -> tuple[pd.DataFrame, pd.DataFrame]:
analysis_display = mapped.copy()
if selected_books:
analysis_display = analysis_display[analysis_display["sportsbook"].isin(selected_books)]
if market_type == "hr" and min_edge > -0.50:
analysis_display = analysis_display[
analysis_display["edge"].notna() & (analysis_display["edge"] >= min_edge)
]
sort_col_map = {
"EV": "bet_ev",
"Edge": "edge",
"Confidence": "confidence_score",
"Pregame Probability": "model_hr_prob",
"Implied Probability": "implied_prob",
}
sort_col = sort_col_map.get(sort_option, "implied_prob")
if sort_col in analysis_display.columns:
analysis_display = analysis_display.sort_values(sort_col, ascending=False, na_position="last")
table_display = analysis_display.copy()
if view == "Best Line":
table_display = select_best_lines_per_prop(table_display)
if sort_col in table_display.columns:
table_display = table_display.sort_values(sort_col, ascending=False, na_position="last")
return analysis_display, table_display
def _modeled_hr_primary_subset(df: pd.DataFrame) -> pd.DataFrame:
if df is None or df.empty:
return pd.DataFrame()
out = df.copy()
market_series = out.get("market_family", out.get("market", pd.Series([""] * len(out), index=out.index)))
market_series = market_series.astype(str).str.strip().str.lower()
threshold_series = pd.to_numeric(out.get("threshold", pd.Series([1] * len(out), index=out.index)), errors="coerce").fillna(1).astype(int)
if "is_modeled" in out.columns:
modeled_series = out["is_modeled"].apply(lambda v: str(v).strip().lower() in {"1", "true", "yes"})
else:
modeled_series = pd.Series([True] * len(out), index=out.index)
if "has_model_probability" in out.columns:
prob_series = out["has_model_probability"].apply(lambda v: str(v).strip().lower() in {"1", "true", "yes"})
else:
prob_series = out.get("model_hr_prob", pd.Series([None] * len(out), index=out.index)).notna()
return out[(market_series == "hr") & (threshold_series == 1) & modeled_series & prob_series].copy()
def _build_hr_health_debug(display: pd.DataFrame, extra_context: dict[str, Any] | None = None) -> dict[str, Any]:
if display is None or display.empty:
out = {
"modeled_hr_rows_total": 0,
"modeled_hr_rows_with_probability": 0,
"modeled_hr_rows_with_edge": 0,
"modeled_hr_rows_missing_probability": 0,
"research_hr_ladder_rows_total": 0,
"health_rows": [],
}
out.update(dict(extra_context or {}))
return out
working = display.copy()
market_series = working.get("market_family", working.get("market", pd.Series([""] * len(working), index=working.index)))
market_series = market_series.astype(str).str.strip().str.lower()
threshold_series = pd.to_numeric(working.get("threshold", pd.Series([1] * len(working), index=working.index)), errors="coerce").fillna(1).astype(int)
modeled_series = (
working["is_modeled"].apply(lambda v: str(v).strip().lower() in {"1", "true", "yes"})
if "is_modeled" in working.columns else
pd.Series([True] * len(working), index=working.index)
)
expected_modeled = (market_series == "hr") & (threshold_series == 1) & modeled_series
has_probability = working.get("has_model_probability", working.get("model_hr_prob", pd.Series([None] * len(working), index=working.index)).notna())
if not isinstance(has_probability, pd.Series):
has_probability = pd.Series(has_probability, index=working.index)
has_probability = has_probability.apply(lambda v: str(v).strip().lower() in {"1", "true", "yes"} if not isinstance(v, bool) else v)
has_edge = working.get("has_modeled_edge", working.get("edge", pd.Series([None] * len(working), index=working.index)).notna())
if not isinstance(has_edge, pd.Series):
has_edge = pd.Series(has_edge, index=working.index)
has_edge = has_edge.apply(lambda v: str(v).strip().lower() in {"1", "true", "yes"} if not isinstance(v, bool) else v)
research_ladders = (market_series == "hr") & (threshold_series > 1)
health_cols = [
"player_name",
"player_name_raw",
"display_label",
"threshold",
"is_modeled",
"model_hr_prob",
"edge",
"bet_ev",
"implied_prob",
"baseline_mode",
"projected_home_pitcher",
"projected_away_pitcher",
"projected_starter_available",
"projected_home_pitcher_source",
"projected_away_pitcher_source",
"starter_cache_source",
"fallback_used",
"projected_starter_match_status",
"resolved_pitcher_name",
"resolved_pitcher_source",
"model_probability_status",
"modeled_row_available",
"modeled_row_missing_reason",
"telemetry_path_status",
"hr_model_tier",
"applied_layers",
"skipped_layers",
"pitcher_resolution_status",
"zone_status",
"family_zone_status",
"arsenal_status",
]
health_rows = working[expected_modeled][[c for c in health_cols if c in working.columns]].copy()
out = {
"modeled_hr_rows_total": int(expected_modeled.sum()),
"modeled_hr_rows_with_probability": int((expected_modeled & has_probability).sum()),
"modeled_hr_rows_with_edge": int((expected_modeled & has_edge).sum()),
"modeled_hr_rows_missing_probability": int((expected_modeled & ~has_probability).sum()),
"research_hr_ladder_rows_total": int(research_ladders.sum()),
"health_rows": health_rows.to_dict("records"),
}
out.update(dict(extra_context or {}))
return out
def _build_shared_component_debug(display: pd.DataFrame, market_type: str) -> dict[str, Any]:
if display is None or display.empty:
return {"market_type": market_type, "rows": [], "executed_rows": [], "gating_rows": [], "failure_summary": []}
cols = [
"market_family",
"player_name",
"player_name_raw",
"display_label",
"projected_home_pitcher",
"projected_away_pitcher",
"projected_home_pitcher_source",
"projected_away_pitcher_source",
"starter_cache_source",
"fallback_used",
"projected_starter_match_status",
"resolved_pitcher_name",
"formula_version",
"shared_matchup_available",
"telemetry_path_status",
"hr_model_tier",
"matchup_coverage_confidence",
"damage_zone_alignment_subscore",
"pitch_mix_exposure_subscore",
"tunnel_damage_subscore",
"count_pattern_damage_subscore",
"handedness_damage_subscore",
"arsenal_fit_subscore",
"environment_amplification_subscore",
"hr_opportunity_projection",
"projected_pitch_count",
"projected_batters_faced",
"projected_innings",
"pitches_per_bf",
"opportunity_confidence",
"opportunity_reasons",
"projected_k_rate",
"expected_strikeouts",
"zone_matchup_subscore",
"family_zone_matchup_subscore",
"tunneling_subscore",
"release_consistency_subscore",
"sequencing_subscore",
"count_leverage_subscore",
"leash_risk_subscore",
"role_certainty_score",
"times_through_order_penalty",
"variance_band_low",
"variance_band_high",
"expected_pitch_family_mix",
"predicted_attack_regions",
"predicted_damage_regions",
"predicted_whiff_regions",
"component_source_map",
"expected_pitch_family_mix",
"baseline_mode",
"pitcher_resolution_status",
"modeled_row_available",
"modeled_row_missing_reason",
]
working = display[[c for c in cols if c in display.columns]].copy()
def _status(row: pd.Series) -> str:
baseline_mode = str(row.get("baseline_mode") or "").strip().lower()
starter_status = str(row.get("projected_starter_match_status") or "").strip().lower()
pitcher_status = str(row.get("pitcher_resolution_status") or "").strip().lower()
shared_available = str(row.get("shared_matchup_available") or "").strip().lower()
telemetry_status = str(row.get("telemetry_path_status") or "").strip().lower()
if not baseline_mode or baseline_mode in {"none", "nan", "unavailable"}:
return "missing_baseline"
if starter_status == "projected_starter_unavailable":
return "projected_starter_unavailable"
if starter_status == "projected_starter_available_but_unresolved":
return "projected_starter_available_but_unresolved"
if pitcher_status in {"pitcher_missing", "unresolved", "matchup_incomplete", "resolved_no_pitcher_statcast"}:
return pitcher_status or "pitcher_resolution_failure"
if shared_available in {"1", "true", "yes"} or telemetry_status in {
"full_telemetry",
"partial_telemetry",
"core_baseline_plus_projected_pitcher",
"baseline_only_degraded",
}:
return "executed"
component_cols = [
"damage_zone_alignment_subscore",
"pitch_mix_exposure_subscore",
"tunnel_damage_subscore",
"count_pattern_damage_subscore",
"handedness_damage_subscore",
"arsenal_fit_subscore",
"zone_matchup_subscore",
"family_zone_matchup_subscore",
"tunneling_subscore",
"sequencing_subscore",
]
if any(pd.notna(row.get(col)) for col in component_cols):
return "executed"
return "prerequisites_not_met"
working["component_execution_status"] = working.apply(_status, axis=1)
executed = working[working["component_execution_status"] == "executed"].copy()
gating = working[working["component_execution_status"] != "executed"].copy()
failure_summary = (
working["component_execution_status"]
.value_counts(dropna=False)
.rename_axis("failure_reason")
.reset_index(name="row_count")
)
return {
"market_type": market_type,
"rows": working.head(30).to_dict("records"),
"executed_rows": executed.head(30).to_dict("records"),
"gating_rows": gating.head(30).to_dict("records"),
"failure_summary": failure_summary.to_dict("records"),
}
def render_props_hero(display_df: pd.DataFrame, view_model: dict[str, Any] | None = None) -> None:
st.markdown(
"""
Pregame Slate
Props Command Center
Actionable pregame props with model probability, edge, EV, confidence, and explainability layered directly onto the slate.
HR is fully modeled today | Strikeouts are live | Alternate ladders stay in research mode
""",
unsafe_allow_html=True,
)
if display_df.empty:
return
games_summary_df = pd.DataFrame() if not view_model else view_model.get("games_summary_df", pd.DataFrame())
featured_props_df = pd.DataFrame() if not view_model else view_model.get("featured_props_df", pd.DataFrame())
available_books = sorted(display_df["sportsbook"].dropna().astype(str).unique().tolist()) if "sportsbook" in display_df.columns else []
modeled_display = _modeled_hr_primary_subset(display_df)
best_edge = modeled_display["edge"].dropna().max() if "edge" in modeled_display.columns and modeled_display["edge"].notna().any() else None
best_ev = modeled_display["bet_ev"].dropna().max() if "bet_ev" in modeled_display.columns and modeled_display["bet_ev"].notna().any() else None
if not modeled_display.empty:
modeled_display = select_best_lines_per_prop(modeled_display)
avg_featured_edge = (
featured_props_df["edge"].dropna().mean()
if not featured_props_df.empty and "edge" in featured_props_df.columns and featured_props_df["edge"].notna().any()
else None
)
avg_featured_ev = (
featured_props_df["bet_ev"].dropna().mean()
if not featured_props_df.empty and "bet_ev" in featured_props_df.columns and featured_props_df["bet_ev"].notna().any()
else None
)
with st.container():
hero_cols = st.columns(6)
hero_cols[0].metric("Games", int(len(games_summary_df)) if not games_summary_df.empty else int(display_df["event_id"].nunique() if "event_id" in display_df.columns else 0))
hero_cols[1].metric("Books", len(available_books))
hero_cols[2].metric("Modeled 1+ HR Rows", int(len(modeled_display)) if not modeled_display.empty else 0)
hero_cols[3].metric("Best Edge (modeled)", _format_edge(float(best_edge)) if best_edge is not None else "-")
hero_cols[4].metric("Best EV (modeled)", _format_ev(float(best_ev)) if best_ev is not None else "-")
hero_cols[5].metric("Avg Featured EV", _format_ev(float(avg_featured_ev)) if avg_featured_ev is not None else "-")
def render_featured_hr_cards(featured_df: pd.DataFrame) -> None:
st.markdown('Featured Props
', unsafe_allow_html=True)
st.markdown("#### Best-Value 1+ HR Plays")
if featured_df.empty:
st.info("No modeled 1+ HR props match the current filters.")
return
card_count = min(4, len(featured_df))
cols = st.columns(card_count)
for idx, (_, row) in enumerate(featured_df.head(card_count).iterrows()):
col = cols[idx % card_count]
with col:
player = str(row.get("player_name_raw") or row.get("player_name") or "-")
matchup = _build_matchup(row)
badge_class = "props-badge success" if idx == 0 else "props-badge"
badge_text = "Top Pick" if idx == 0 else "Featured"
verdict = str(row.get("verdict") or "tracked").strip().lower()
edge_class = _metric_tone_class("edge", row.get("edge"))
ev_class = _metric_tone_class("ev", row.get("bet_ev"))
model_voice = str(row.get("model_voice") or "Model voice is still being assembled for this matchup.")
line_label = _display_market_line_label(row)
st.markdown(
f"""
{badge_text}
{_render_verdict_badge(verdict)}
{player}
{matchup}
{line_label} | {str(row.get('sportsbook') or '-')}
Odds
{_format_odds(row.get('odds_american'))}
EV
{_format_ev(row.get('bet_ev'))}
Edge
{_format_edge(row.get('edge'))}
Pregame HR%
{_format_pct(row.get('model_hr_prob'))}
Implied
{_format_pct(row.get('implied_prob'))}
Confidence
{_build_confidence_metric_html(row)}
Model Voice: {model_voice}
""",
unsafe_allow_html=True,
)
def render_best_on_slate_cards(best_df: pd.DataFrame, summary: dict[str, Any] | None = None) -> None:
st.markdown('Slate-Wide Best Value
', unsafe_allow_html=True)
st.markdown("#### Best Value Available")
st.caption("Compares the best currently modeled 1+ HR and pitcher strikeout lines across books, then ranks the strongest bettor-value spots on the slate.")
summary = summary or {}
metric_cols = st.columns(4)
metric_cols[0].metric("Modeled Props", int(summary.get("modeled_props_count") or 0))
metric_cols[1].metric("Books Active", int(summary.get("sportsbooks_count") or 0))
metric_cols[2].metric("Markets Active", int(summary.get("markets_count") or 0))
best_ev = summary.get("best_ev")
metric_cols[3].metric("Best EV", _format_ev(float(best_ev)) if best_ev is not None else "-")
if best_df is None or best_df.empty:
st.info("No slate-wide modeled props are currently available.")
return
card_count = min(4, len(best_df))
cols = st.columns(card_count)
for idx, (_, row) in enumerate(best_df.head(card_count).iterrows()):
col = cols[idx % card_count]
with col:
player = str(row.get("player_name_raw") or row.get("player_name") or "-")
matchup = _build_matchup(row)
badge_class = "props-badge success" if idx == 0 else "props-badge"
badge_text = "Best On Slate" if idx == 0 else "Slate Value"
verdict = str(row.get("verdict") or "tracked").strip().lower()
edge_class = _metric_tone_class("edge", row.get("edge"))
ev_class = _metric_tone_class("ev", row.get("bet_ev"))
model_voice = str(row.get("model_voice") or "Model voice is still being assembled for this matchup.")
market_family = str(row.get("market_family") or row.get("market") or "").strip().lower()
probability_label = "Pregame HR%" if market_family == "hr" else "Fair%"
probability_value = row.get("model_hr_prob") if market_family == "hr" else row.get("fair_prob")
market_badge = _display_market_line_label(row)
st.markdown(
f"""
{badge_text}
{_render_verdict_badge(verdict)}
{player}
{matchup}
{market_badge} | {str(row.get('sportsbook') or '-')}
Odds
{_format_odds(row.get('odds_american'))}
EV
{_format_ev(row.get('bet_ev'))}
Edge
{_format_edge(row.get('edge'))}
{probability_label}
{_format_pct(probability_value)}
Implied
{_format_pct(row.get('implied_prob'))}
Confidence
{_build_confidence_metric_html(row)}
Model Voice: {model_voice}
""",
unsafe_allow_html=True,
)
def _build_terminal_top_bets_df(display: pd.DataFrame, market_type: str, limit: int = 7) -> pd.DataFrame:
if display is None or display.empty:
return pd.DataFrame()
working = display.copy()
if market_type == "hr":
working = _modeled_hr_primary_subset(working)
elif "is_modeled" in working.columns:
working = working[working["is_modeled"] == True].copy()
if working.empty:
return pd.DataFrame()
working = select_best_lines_per_prop(working)
sort_cols: list[str] = []
ascending: list[bool] = []
for col in ("bet_ev", "edge", "confidence_score", "final_recommendation_score"):
if col in working.columns:
sort_cols.append(col)
ascending.append(False)
if "odds_american" in working.columns:
sort_cols.append("odds_american")
ascending.append(False)
if sort_cols:
working = working.sort_values(sort_cols, ascending=ascending, na_position="last")
return working.head(max(1, int(limit))).reset_index(drop=True)
def _render_terminal_header(display: pd.DataFrame, market_type: str, top_bets_df: pd.DataFrame) -> None:
market_label = _market_label(market_type)
available_books = (
sorted(display["sportsbook"].dropna().astype(str).unique().tolist())
if display is not None and not display.empty and "sportsbook" in display.columns
else []
)
best_edge = (
pd.to_numeric(top_bets_df.get("edge"), errors="coerce").dropna().max()
if not top_bets_df.empty and "edge" in top_bets_df.columns
else None
)
best_ev = (
pd.to_numeric(top_bets_df.get("bet_ev"), errors="coerce").dropna().max()
if not top_bets_df.empty and "bet_ev" in top_bets_df.columns
else None
)
st.markdown(
f"""
""",
unsafe_allow_html=True,
)
metrics = st.columns(5)
metrics[0].metric("Games", int(display["event_id"].nunique()) if display is not None and not display.empty and "event_id" in display.columns else 0)
metrics[1].metric("Books", len(available_books))
metrics[2].metric("Top Bets", int(len(top_bets_df)))
metrics[3].metric("Best Edge", _format_edge(float(best_edge)) if best_edge is not None else "-")
metrics[4].metric("Best EV", _format_ev(float(best_ev)) if best_ev is not None else "-")
def _render_top_bet_highlight(row: pd.Series | dict[str, Any], rank_label: str) -> None:
player = str(row.get("player_name_raw") or row.get("player_name") or "-")
matchup = _build_matchup(row)
market_line = _display_market_line_label(row)
book = str(row.get("sportsbook") or "-")
model_voice = str(row.get("model_voice") or "No model voice available.")
market_family = str(row.get("market_family") or row.get("market") or "").strip().lower()
probability_value = row.get("model_hr_prob") if market_family == "hr" else row.get("fair_prob")
probability_label = "Pregame HR%" if market_family == "hr" else "Fair%"
edge_class = _metric_tone_class("edge", row.get("edge"))
ev_class = _metric_tone_class("ev", row.get("bet_ev"))
st.markdown(
f"""
{rank_label}
{player}
{matchup}
{market_line} | {book}
Odds
{_format_odds(row.get('odds_american'))}
EV
{_format_ev(row.get('bet_ev'))}
Edge
{_format_edge(row.get('edge'))}
{probability_label}
{_format_pct(probability_value)}
Implied
{_format_pct(row.get('implied_prob'))}
Confidence
{_format_confidence(row.get('confidence_score'))}
Thesis: {model_voice}
""",
unsafe_allow_html=True,
)
def _render_terminal_mini_card(row: pd.Series | dict[str, Any], rank: int) -> None:
player = str(row.get("player_name_raw") or row.get("player_name") or "-")
matchup = _build_matchup(row)
market_line = _display_market_line_label(row)
book = str(row.get("sportsbook") or "-")
model_voice = str(row.get("model_voice") or "No thesis available.")
model_voice = model_voice[:140] + ("..." if len(model_voice) > 140 else "")
edge_class = _metric_tone_class("edge", row.get("edge"))
ev_class = _metric_tone_class("ev", row.get("bet_ev"))
st.markdown(
f"""
#{rank}
{player}
{matchup}
{market_line} | {book}
EV
{_format_ev(row.get('bet_ev'))}
Edge
{_format_edge(row.get('edge'))}
Odds
{_format_odds(row.get('odds_american'))}
Conf
{_format_confidence(row.get('confidence_score'))}
{model_voice}
""",
unsafe_allow_html=True,
)
def render_terminal_top_bets_board(top_bets_df: pd.DataFrame, market_type: str) -> None:
st.markdown('Top Bets
', unsafe_allow_html=True)
st.markdown("#### Best Current Value On The Slate")
if top_bets_df is None or top_bets_df.empty:
st.info(f"No top {_market_label(market_type)} bets are available for the current filters.")
return
st.markdown('', unsafe_allow_html=True)
_render_top_bet_highlight(top_bets_df.iloc[0], "Top Play")
if len(top_bets_df) > 1:
remaining = top_bets_df.iloc[1:].reset_index(drop=True)
cols_per_row = 3
for start in range(0, len(remaining), cols_per_row):
cols = st.columns(min(cols_per_row, len(remaining) - start))
for idx, (_, row) in enumerate(remaining.iloc[start:start + cols_per_row].iterrows(), start=start + 2):
with cols[idx - start - 2]:
_render_terminal_mini_card(row, idx)
st.markdown("
", unsafe_allow_html=True)
def _build_generic_game_workspace(display: pd.DataFrame) -> tuple[pd.DataFrame, dict[str, Any]]:
if display is None or display.empty:
return pd.DataFrame(), {}
working = select_best_lines_per_prop(display.copy())
summary_rows: list[dict[str, Any]] = []
game_map: dict[str, Any] = {}
for (event_id, away_team, home_team, commence_time), game_df in working.groupby(
["event_id", "away_team", "home_team", "commence_time"],
dropna=False,
):
game_key = str(event_id or f"{away_team}|{home_team}|{commence_time}")
sort_df = game_df.copy()
if "bet_ev" in sort_df.columns:
sort_df = sort_df.sort_values(["bet_ev", "edge", "confidence_score"], ascending=[False, False, False], na_position="last")
top_row = sort_df.iloc[0].to_dict() if not sort_df.empty else {}
player_rows: list[dict[str, Any]] = []
for player_name, player_df in sort_df.groupby("player_name", dropna=False):
primary_row = player_df.iloc[0].to_dict()
player_rows.append(
{
"player_key": f"{game_key}|{str(primary_row.get('player_name') or '').strip().lower()}",
"player_name": primary_row.get("player_name"),
"player_name_raw": primary_row.get("player_name_raw"),
"best_display_label": primary_row.get("display_label"),
"best_book": primary_row.get("sportsbook"),
"best_odds_american": primary_row.get("odds_american"),
"best_edge": primary_row.get("edge"),
"best_bet_ev": primary_row.get("bet_ev"),
"best_confidence_score": primary_row.get("confidence_score"),
"best_verdict": primary_row.get("verdict"),
"best_model_hr_prob": primary_row.get("fair_prob"),
"model_voice": primary_row.get("model_voice"),
"model_voice_primary_reason": primary_row.get("model_voice_primary_reason"),
"model_voice_caveat": primary_row.get("model_voice_caveat"),
"details": {
"best_primary_row": primary_row,
"primary_rows": player_df.to_dict("records"),
"alt_rows": [],
},
}
)
game_map[game_key] = {
"game_key": game_key,
"event_id": event_id,
"away_team": away_team,
"home_team": home_team,
"commence_time": commence_time,
"modeled_props_count": int(len(sort_df)),
"players_count": int(game_df["player_name"].nunique()),
"best_edge": top_row.get("edge"),
"best_bet_ev": top_row.get("bet_ev"),
"top_player_name": top_row.get("player_name"),
"top_display_label": top_row.get("display_label"),
"top_book": top_row.get("sportsbook"),
"top_verdict": top_row.get("verdict"),
"players": player_rows,
}
summary_rows.append({key: value for key, value in game_map[game_key].items() if key != "players"})
summary_df = pd.DataFrame(summary_rows)
if not summary_df.empty and "best_edge" in summary_df.columns:
summary_df = summary_df.sort_values(["best_edge", "best_bet_ev"], ascending=[False, False], na_position="last").reset_index(drop=True)
return summary_df, game_map
def _normalize_game_summary_rows(
market_type: str,
analysis_display: pd.DataFrame,
view_model: dict[str, Any] | None = None,
) -> tuple[pd.DataFrame, dict[str, Any]]:
if market_type == "hr" and isinstance(view_model, dict):
return (
view_model.get("games_summary_df", pd.DataFrame()),
view_model.get("game_player_props_map", {}),
)
return _build_generic_game_workspace(analysis_display)
def render_game_navigator_terminal(game_summary_df: pd.DataFrame, market_type: str) -> str | None:
st.markdown('Game Navigator
', unsafe_allow_html=True)
st.markdown("#### Jump Into A Matchup")
if game_summary_df is None or game_summary_df.empty:
st.info(f"No {_market_label(market_type)} games are available for navigation.")
return None
state_key = f"props_selected_game_{market_type}"
ordered_game_keys = game_summary_df["game_key"].astype(str).tolist()
current = st.session_state.get(state_key)
if current not in ordered_game_keys:
current = ordered_game_keys[0]
st.session_state[state_key] = current
st.markdown('', unsafe_allow_html=True)
cols_per_row = 3
for start in range(0, len(game_summary_df), cols_per_row):
row_df = game_summary_df.iloc[start:start + cols_per_row]
cols = st.columns(len(row_df))
for col, (_, row) in zip(cols, row_df.iterrows()):
game_key = str(row.get("game_key") or "")
selected_class = " selected" if game_key == current else ""
with col:
st.markdown(
f"""
{_build_matchup(row)}
{_build_game_time(row)}
{int(row.get('modeled_props_count') or 0)} modeled props | best edge {_format_edge(row.get('best_edge'))}
top: {str(row.get('top_player_name') or '-')} | {str(row.get('top_display_label') or '-')}
""",
unsafe_allow_html=True,
)
if st.button("View Game", key=f"props_game_nav_{market_type}_{game_key}", use_container_width=True):
st.session_state[state_key] = game_key
current = game_key
st.markdown("
Selected Game
', unsafe_allow_html=True)
st.markdown("#### Prop Workspace")
if not selected_game_key or selected_game_key not in game_map:
st.info("Select a game to inspect player props and books.")
return
game_payload = game_map[selected_game_key]
st.markdown(
f"""
{_build_matchup(game_payload)}
{_build_game_time(game_payload)} | {int(game_payload.get('modeled_props_count') or 0)} modeled props | best edge {_format_edge(game_payload.get('best_edge'))} | best EV {_format_ev(game_payload.get('best_bet_ev'))}
""",
unsafe_allow_html=True,
)
player_entries = game_payload.get("players") or []
if not player_entries:
st.info("No player props available for this game.")
return
for player_entry in player_entries:
render_player_hr_row(player_entry)
def render_player_hr_details(player_details: dict[str, Any]) -> None:
primary_rows = pd.DataFrame(player_details.get("primary_rows") or [])
alt_rows = pd.DataFrame(player_details.get("alt_rows") or [])
primary_market_family = str((player_details.get("best_primary_row") or {}).get("market_family") or "").strip().lower()
if not primary_rows.empty:
st.caption("All books - 1+ HR" if primary_market_family in {"", "hr"} else f"All books - {_market_label(primary_market_family)}")
primary_display_cols = [
"sportsbook", "display_label", "odds_american", "implied_prob",
"model_hr_prob", "fair_prob", "bet_ev", "edge", "confidence_score", "verdict", "model_hr_prob_source",
]
primary_display = primary_rows[[c for c in primary_display_cols if c in primary_rows.columns]].copy()
if "odds_american" in primary_display.columns:
primary_display["odds_american"] = primary_display["odds_american"].apply(_format_odds)
if "implied_prob" in primary_display.columns:
primary_display["implied_prob"] = primary_display["implied_prob"].apply(_format_pct)
if "model_hr_prob" in primary_display.columns:
primary_display["model_hr_prob"] = primary_display["model_hr_prob"].apply(_format_pct)
if "fair_prob" in primary_display.columns:
primary_display["fair_prob"] = primary_display["fair_prob"].apply(_format_pct)
if "bet_ev" in primary_display.columns:
primary_display["bet_ev"] = primary_display["bet_ev"].apply(_format_ev)
if "edge" in primary_display.columns:
primary_display["edge"] = primary_display["edge"].apply(_format_edge)
if "confidence_score" in primary_display.columns:
primary_display["confidence_score"] = primary_display["confidence_score"].apply(_format_confidence)
st.dataframe(_style_metric_dataframe(primary_display), use_container_width=True, hide_index=True)
if not alt_rows.empty:
st.caption("Alternate HR ladders")
alt_display_cols = [
"sportsbook", "display_label", "odds_american", "implied_prob",
"model_hr_prob_source",
]
alt_display = alt_rows[[c for c in alt_display_cols if c in alt_rows.columns]].copy()
if "odds_american" in alt_display.columns:
alt_display["odds_american"] = alt_display["odds_american"].apply(_format_odds)
if "implied_prob" in alt_display.columns:
alt_display["implied_prob"] = alt_display["implied_prob"].apply(_format_pct)
st.dataframe(alt_display, use_container_width=True, hide_index=True)
st.markdown(
'Alternate HR ladders are tracked for price discovery, but only 1+ HR is modeled in this release.
',
unsafe_allow_html=True,
)
best_primary = player_details.get("best_primary_row") or {}
diagnostic_fields = [
"resolved_pitcher_name",
"baseline_hr_prob",
"raw_hr_prob",
"calibrated_hr_prob",
"pregame_hr_prob",
"fair_prob",
"bet_ev",
"verdict",
"model_voice",
"model_voice_primary_reason",
"model_voice_caveat",
"model_voice_reason_candidates",
"model_voice_tags",
"model_voice_for",
"model_voice_against",
"confidence_score",
"confidence_score_raw",
"confidence_score_display",
"confidence_source",
"confidence_bucket",
"confidence_reasons",
"confidence_component_bonuses",
"confidence_component_penalties",
"confidence_primary_driver",
"confidence_summary_label",
"opportunity_hr_adjustment",
"expected_pa",
"lineup_slot_used",
"lineup_slot_source",
"team_total_used",
"projected_home_pitcher",
"projected_away_pitcher",
"projected_starter_available",
"projected_home_pitcher_source",
"projected_away_pitcher_source",
"starter_cache_source",
"fallback_used",
"projected_starter_match_status",
"resolved_pitcher_source",
"telemetry_path_status",
"hr_model_tier",
"shared_matchup_available",
"modeled_row_available",
"modeled_row_missing_reason",
"pitcher_hr_adjustment",
"env_hr_adjustment",
"park_hr_adjustment",
"weather_hr_adjustment",
"rolling_hr_adjustment",
"applied_layers",
"skipped_layers",
"pitcher_swstr_rate",
"pitcher_csw_rate",
"pitcher_ball_rate",
"arsenal_whiff_risk",
"family_zone_whiff_risk",
"zone_whiff_risk",
"expected_strikeouts",
"pitches_per_bf",
"opportunity_confidence",
"opportunity_reasons",
"sequencing_score",
"trajectory_tunnel_score",
"trajectory_release_consistency_score",
]
diag_payload = {field: best_primary.get(field) for field in diagnostic_fields if field in best_primary}
if diag_payload:
st.caption("Primary-line diagnostics")
st.json(diag_payload, expanded=False)
def render_player_hr_row(player_entry: dict[str, Any]) -> None:
player_name = str(player_entry.get("player_name_raw") or player_entry.get("player_name") or "-")
best_label = str(player_entry.get("best_display_label") or "1+ HR")
best_book = str(player_entry.get("best_book") or "-")
best_odds = _format_odds(player_entry.get("best_odds_american"))
best_edge = _format_edge(player_entry.get("best_edge"))
best_ev = _format_ev(player_entry.get("best_bet_ev"))
best_confidence = _format_confidence(player_entry.get("best_confidence_score"))
verdict = str(player_entry.get("best_verdict") or "tracked").strip().lower()
model_voice = str(player_entry.get("model_voice") or "")
details_label = f"{player_name} | {best_label} | {best_book} {best_odds} | {best_ev} EV | {best_edge} | {best_confidence} conf"
if player_entry.get("has_alt_ladders"):
details_label += " | alt ladders"
with st.container():
st.markdown(f"**{player_name}**")
st.markdown(_render_verdict_badge(verdict), unsafe_allow_html=True)
summary_cols = st.columns([0.8, 0.9, 0.7, 0.7, 0.7, 1.1])
summary_cols[0].caption(best_label)
summary_cols[1].caption(f"{best_book} {best_odds}")
summary_cols[2].caption(f"EV {best_ev}")
summary_cols[3].caption(f"Edge {best_edge}")
summary_cols[4].caption(f"Conf {best_confidence}")
details_open = summary_cols[5].toggle(
"Show books and ladders",
value=False,
key=f"props_player_toggle_{player_entry.get('player_key')}",
)
if model_voice:
st.caption(f"Model Voice: {model_voice}")
if details_open:
st.caption(details_label)
details = player_entry.get("details") or {}
metric_cols = st.columns(6)
metric_cols[0].metric("Best Book", best_book)
metric_cols[1].metric("Best Odds", best_odds)
metric_cols[2].metric(
"Pregame HR%" if str((details.get("best_primary_row") or {}).get("market_family") or "hr").strip().lower() == "hr" else "Fair%",
_format_pct(player_entry.get("best_model_hr_prob")),
)
metric_cols[3].metric("EV", best_ev)
metric_cols[4].metric("Edge", best_edge)
metric_cols[5].metric("Confidence", best_confidence)
if model_voice:
st.caption(f"Model Voice: {model_voice}")
primary_reason = str(player_entry.get("model_voice_primary_reason") or "")
caveat = str(player_entry.get("model_voice_caveat") or "")
if primary_reason or caveat:
why_lines: list[str] = []
if primary_reason:
why_lines.append(f"Lead reason: {primary_reason}")
if caveat:
why_lines.append(f"Caveat: {caveat}")
st.caption("Why this rating")
for line in why_lines:
st.write(f"- {line}")
_render_confidence_breakdown(details.get("best_primary_row") or {})
render_player_hr_details(details)
st.divider()
def render_game_card(game_payload: dict[str, Any]) -> None:
matchup = _build_matchup(game_payload)
commence_time = _build_game_time(game_payload)
best_edge = _format_edge(game_payload.get("best_edge"))
top_player = str(game_payload.get("top_player_name") or "-")
top_label = str(game_payload.get("top_display_label") or "-")
top_book = str(game_payload.get("top_book") or "-")
top_verdict = str(game_payload.get("top_verdict") or "tracked").upper()
st.markdown(
f"""
{matchup}
{commence_time}
{int(game_payload.get('modeled_props_count') or 0)} modeled props | best edge {best_edge} | best EV {_format_ev(game_payload.get('best_bet_ev'))}
Best current prop: {top_player} | {top_label} | {top_book} | {top_verdict}
""",
unsafe_allow_html=True,
)
label = f"Open {matchup}"
with st.expander(label, expanded=False):
player_entries = game_payload.get("players") or []
if not player_entries:
st.info("No player props available for this game.")
return
for player_entry in player_entries:
render_player_hr_row(player_entry)
def render_game_explorer(game_player_props_map: dict[str, dict[str, Any]]) -> None:
st.markdown('', unsafe_allow_html=True)
st.markdown('Research Board
', unsafe_allow_html=True)
st.markdown("#### By-Game Explorer")
st.caption("Open a matchup to compare each player's best 1+ HR line, then expand a player for all books and alternate ladders.")
if not game_player_props_map:
st.info("No game-grouped props available for the current filters.")
return
for game_payload in game_player_props_map.values():
render_game_card(game_payload)
def _build_flat_table_rows(display: pd.DataFrame, market_type: str) -> list[dict[str, Any]]:
if market_type == "hr":
return [
{
"Player": str(row.get("player_name_raw") or row.get("player_name") or "-"),
"Matchup": _build_matchup(row),
"Game Time": _build_game_time(row),
"Book": str(row.get("sportsbook") or "-"),
"HR Line": _display_market_line_label(row),
"Verdict": str(row.get("verdict") or "-").upper(),
"Odds": _format_odds(row.get("odds_american")),
"Implied%": _format_pct(row.get("implied_prob")),
"Pregame HR%": _format_pct(row.get("model_hr_prob")),
"EV": _format_ev(row.get("bet_ev")),
"Confidence": _format_confidence(row.get("confidence_score")),
"Model Voice": str(row.get("model_voice") or "-"),
"Source": str(row.get("model_hr_prob_source") or "-"),
"Edge": _format_edge(row.get("edge")),
}
for _, row in display.iterrows()
]
return [
{
"Player": str(row.get("player_name_raw") or row.get("player_name") or "-"),
"Matchup": _build_matchup(row),
"Game Time": _build_game_time(row),
"Book": str(row.get("sportsbook") or "-"),
"Market": _display_market_line_label(row),
"Verdict": str(row.get("verdict") or "-").upper(),
"Odds": _format_odds(row.get("odds_american")),
"Implied%": _format_pct(row.get("implied_prob")),
"Fair%": _format_pct(row.get("fair_prob")),
"EV": _format_ev(row.get("bet_ev")),
"Confidence": _format_confidence(row.get("confidence_score")),
"Model Voice": str(row.get("model_voice") or "-"),
"Edge": _format_edge(row.get("edge")),
}
for _, row in display.iterrows()
]
def render_flat_props_table(display: pd.DataFrame, market_type: str) -> None:
st.markdown("#### Flat Props Table")
if market_type == "no_hr":
st.caption("No Home Run is tracked when available, but it is not actionable until a fair-probability model is active.")
table_df = pd.DataFrame(_build_flat_table_rows(display, market_type))
st.dataframe(_style_metric_dataframe(table_df), use_container_width=True, hide_index=True)
def _render_summary_metrics(display: pd.DataFrame, market_type: str) -> None:
if market_type == "hr":
col1, col2, col3 = st.columns(3)
modeled_display = _modeled_hr_primary_subset(display)
col1.metric("Shown modeled 1+ HR", len(modeled_display))
with_edge = modeled_display["edge"].dropna() if "edge" in modeled_display.columns else pd.Series(dtype=float)
with_ev = modeled_display["bet_ev"].dropna() if "bet_ev" in modeled_display.columns else pd.Series(dtype=float)
col2.metric("Shown rows with priced edge", len(with_edge))
col3.metric("Best EV", _format_ev(float(with_ev.max())) if not with_ev.empty else "-")
else:
col1, col2, col3 = st.columns(3)
col1.metric("Props shown", len(display))
modeled = display[display["is_modeled"] == True] if "is_modeled" in display.columns else pd.DataFrame()
col2.metric("Modeled", len(modeled))
with_ev = display["bet_ev"].dropna() if "bet_ev" in display.columns else pd.Series(dtype=float)
col3.metric("Best EV", _format_ev(float(with_ev.max())) if not with_ev.empty else "-")
def render_probability_diagnostics(display: pd.DataFrame) -> None:
with st.expander("Probability Diagnostics", expanded=False):
diag_cols = [
"player_name",
"sportsbook",
"display_label",
"verdict",
"model_voice",
"model_voice_primary_reason",
"model_voice_caveat",
"model_voice_for",
"model_voice_against",
"baseline_hr_prob",
"raw_hr_prob",
"calibrated_hr_prob",
"pregame_hr_prob",
"fair_prob",
"bet_ev",
"confidence_score",
"confidence_bucket",
"confidence_reasons",
"opportunity_hr_adjustment",
"expected_pa",
"lineup_slot_used",
"lineup_slot_source",
"team_total_used",
"pitcher_hr_adjustment",
"trend_hr_adjustment",
"zone_hr_adjustment",
"family_zone_hr_adjustment",
"arsenal_hr_adjustment",
"pulled_contact_hr_adjustment",
"env_hr_adjustment",
"park_hr_adjustment",
"weather_hr_adjustment",
"platoon_hr_adjustment",
"trajectory_hr_adjustment",
"rolling_hr_adjustment",
"expected_strikeouts",
"pitcher_swstr_rate",
"pitcher_csw_rate",
"pitcher_ball_rate",
"arsenal_whiff_risk",
"family_zone_whiff_risk",
"zone_whiff_risk",
"sequencing_score",
"trajectory_tunnel_score",
"trajectory_release_consistency_score",
"applied_layers",
"skipped_layers",
"resolved_pitcher_name",
]
diag_display = display[[c for c in diag_cols if c in display.columns]].copy()
st.dataframe(diag_display, use_container_width=True, hide_index=True)
def render_execution_layer(display: pd.DataFrame) -> None:
if "final_recommendation_score" not in display.columns:
return
with st.expander("Execution Layer", expanded=False):
exec_cols = [
"player_name",
"sportsbook",
"display_label",
"edge_raw",
"edge_filtered",
"execution_confidence_score",
"execution_volatility_score",
"execution_signal_strength_score",
"market_width",
"market_outlier_flag",
"stale_book_flag",
"timing_flag",
"correlation_flag",
"final_recommendation_score",
"edge_filter_flags",
]
exec_display = display[[c for c in exec_cols if c in display.columns]].copy()
exec_display = exec_display.sort_values(
"final_recommendation_score",
ascending=False,
na_position="last",
)
st.dataframe(exec_display, use_container_width=True, hide_index=True)
def render_props(
statcast_df: pd.DataFrame | None = None,
conn=None,
raw_props: pd.DataFrame | None = None,
pitcher_statcast_df: pd.DataFrame | None = None,
probable_starters: dict | None = None,
) -> None:
_render_props_ui_styles()
raw = _load_raw_props(raw_props)
if raw.empty:
render_props_hero(pd.DataFrame(), view_model=None)
_render_empty_props_state()
return
if "market" not in raw.columns:
raw["market"] = "hr"
available_markets = sorted(raw["market"].dropna().unique().tolist())
default_idx = available_markets.index("hr") if "hr" in available_markets else 0
market_type = st.selectbox(
"Market",
options=available_markets,
index=default_idx,
key="props_market",
format_func=_market_label,
)
prepared_bundle = _load_props_prepared_bundle(
raw=raw,
probable_starters=probable_starters,
)
import time as _time
if prepared_bundle.get("snapshot_source_status") in ("runtime_fallback_timeout", "patch_build_timeout"):
st.session_state.setdefault("props_baseline_reload_at", _time.time() + 360)
_reload_at = st.session_state.get("props_baseline_reload_at")
if _reload_at:
if _time.time() < _reload_at:
st.info(
"📊 **Player baseline data is loading in the background.** "
"Props are shown with basic line analysis. "
"The page will refresh automatically with full Statcast enrichment in a few minutes."
)
else:
del st.session_state["props_baseline_reload_at"]
_load_props_prepared_bundle.clear()
st.rerun()
prepared_signature = prepared_bundle.get("signature")
if st.session_state.get("_props_prepared_signature") != prepared_signature:
st.session_state["_props_prepared_signature"] = prepared_signature
st.session_state["props_prepared_bundle"] = prepared_bundle
st.session_state["props_raw_feed"] = raw
st.session_state["props_supported_markets"] = list(prepared_bundle.get("supported_markets") or [])
st.session_state["props_modeled_market_bundle"] = {}
st.session_state.pop("props_view_model_bundle", None)
st.session_state.pop("props_exec_df", None)
st.session_state.pop("props_market_debug_bundle", None)
st.session_state.pop("_props_market_debug_signature", None)
st.session_state.pop("_props_exec_df_signature", None)
st.session_state.pop("_props_exec_df_combined", None)
else:
st.session_state["props_prepared_bundle"] = prepared_bundle
st.session_state["props_raw_feed"] = raw
st.session_state["props_supported_markets"] = list(prepared_bundle.get("supported_markets") or [])
starter_bundle = prepared_bundle.get("starter_bundle") or {}
st.session_state["props_starter_debug"] = {
**dict(prepared_bundle.get("starter_debug") or {}),
"starter_cache_age_seconds": st.session_state.get("probable_starters_cache_age_seconds"),
"starter_refresh_mode": st.session_state.get("probable_starters_refresh_mode"),
}
filtered_raw = raw[raw["market"] == market_type].copy()
if filtered_raw.empty:
render_props_hero(pd.DataFrame(), view_model=None)
st.info(f"No {market_type.upper()} props in current feed.")
return
modeled_market_bundle = st.session_state.get("props_modeled_market_bundle") or {}
market_payload = modeled_market_bundle.get(market_type)
if market_payload is None:
market_payload = _build_market_payload_from_prepared_bundle(
raw=raw,
market_type=market_type,
prepared_bundle=prepared_bundle,
capture_debug=False,
)
if market_payload is not None:
modeled_market_bundle = dict(modeled_market_bundle)
modeled_market_bundle[market_type] = market_payload
st.session_state["props_modeled_market_bundle"] = modeled_market_bundle
if market_payload is None:
render_props_hero(pd.DataFrame(), view_model=None)
st.info("No props could be prepared for the selected market.")
return
mapped = market_payload["mapped"]
st.session_state["props_exec_df"] = mapped
_hydrate_props_debug_state(
market_type=market_type,
payload=market_payload,
)
if mapped.empty:
render_props_hero(pd.DataFrame(), view_model=None)
st.info("No mappable HR prop rows." if market_type == "hr" else "No props available.")
return
_log_t = threading.Thread(target=_maybe_log_props, args=(conn, mapped.copy()), daemon=True)
_log_t.start()
_log_t.join(timeout=5)
view_model = build_hr_props_view_model(mapped) if market_type == "hr" else None
if view_model is not None:
st.session_state["props_view_model_bundle"] = view_model
else:
st.session_state.pop("props_view_model_bundle", None)
selected_books, min_edge, sort_option, view = _get_current_filter_state(mapped, market_type)
analysis_display, table_display = _prepare_display_frames(
mapped=mapped,
market_type=market_type,
selected_books=selected_books,
min_edge=min_edge,
sort_option=sort_option,
view=view,
)
if analysis_display.empty:
_render_terminal_header(mapped, market_type, pd.DataFrame())
st.info("No props match the current filters.")
return
top_bets_df = _build_terminal_top_bets_df(analysis_display, market_type, limit=7)
_render_terminal_header(analysis_display, market_type, top_bets_df)
render_terminal_top_bets_board(top_bets_df, market_type)
st.markdown('', unsafe_allow_html=True)
st.markdown('
Controls
', unsafe_allow_html=True)
st.markdown('
Use the active filters to reshape the board, then jump directly into a matchup workspace below.
', unsafe_allow_html=True)
selected_books, min_edge, sort_option, view = _render_filter_controls(mapped, market_type)
st.markdown("
', unsafe_allow_html=True)
_render_market_coverage_note(mapped, market_type)
st.markdown("
", unsafe_allow_html=True)
analysis_display, table_display = _prepare_display_frames(
mapped=mapped,
market_type=market_type,
selected_books=selected_books,
min_edge=min_edge,
sort_option=sort_option,
view=view,
)
if analysis_display.empty:
st.info("No props match the current filters.")
return
slate_modeled_df = _build_best_on_slate_source(
modeled_market_bundle=modeled_market_bundle,
active_market_type=market_type,
active_mapped=mapped,
)
best_on_slate_df = build_best_on_slate_df(slate_modeled_df, limit=8)
best_on_slate_summary = build_best_on_slate_summary(slate_modeled_df)
st.session_state["props_best_on_slate_debug"] = {
"rows": best_on_slate_df.to_dict("records") if not best_on_slate_df.empty else [],
"summary": best_on_slate_summary,
}
filtered_view_model = build_hr_props_view_model(analysis_display) if market_type == "hr" else None
game_summary_df, game_map = _normalize_game_summary_rows(
market_type=market_type,
analysis_display=analysis_display,
view_model=filtered_view_model,
)
selected_game_key = render_game_navigator_terminal(game_summary_df, market_type)
render_selected_game_workspace(
selected_game_key=selected_game_key,
game_map=game_map,
market_type=market_type,
)
st.markdown('Research Surfaces
', unsafe_allow_html=True)
bottom_tabs = st.tabs(["Flat Table", "Probability", "Execution", "Legend"])
with bottom_tabs[0]:
render_flat_props_table(table_display, market_type)
with bottom_tabs[1]:
render_probability_diagnostics(analysis_display)
with bottom_tabs[2]:
render_execution_layer(analysis_display)
if market_type == "hr":
st.caption(
"Pregame HR% starts from the batter baseline and applies pitcher, matchup, park/weather, trajectory, and rolling context. "
"Live-only pitch telemetry and count/base-out state remain part of the live Dashboard path."
)
with bottom_tabs[3]:
_render_summary_metrics(table_display, market_type)
_render_props_legend()