import os import json import time import math import copy from dataclasses import dataclass, asdict, field from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from urllib.parse import quote import requests import pandas as pd import gradio as gr # ========================================================= # APP CONFIG # ========================================================= APP_TITLE = "Live UFL / NFL Player Trading Desk" APP_SUBTITLE = ( "Unified multi-market trading workspace with frozen prematch session slate, " "shared game clock, and trader-facing live pricing." ) # Primary endpoints from the current tool. The app is intentionally structured so # additional endpoints/subcategories can be added later without touching the UI layer. EVENTS_API_URL = ( "https://sportsbook-nash.draftkings.com/sites/US-VA-SB/api/" "sportscontent/controldata/league/leagueSubcategory/v1/markets" "?isBatchable=false" "&templateVars=212333%2C4518" "&eventsQuery=%24filter%3DleagueId%20eq%20%27212333%27%20AND%20" "clientMetadata%2FSubcategories%2Fany%28s%3A%20s%2FId%20eq%20%274518%27%29" "&marketsQuery=%24filter%3DclientMetadata%2FsubCategoryId%20eq%20%274518%27%20AND%20" "tags%2Fall%28t%3A%20t%20ne%20%27SportcastBetBuilder%27%29" "&include=Events" "&entity=events" ) # DraftKings subcategory IDs that can be expanded later. # The parser is written to handle partially available markets gracefully. SUBCATEGORY_IDS = { "anytime_td": "13077", # existing known path "passing_yards": "12084", # best-effort placeholders / future extension points "passing_tds": "12080", "rushing_yards": "12083", "receiving_yards": "12085", } HEADERS = { "User-Agent": "Mozilla/5.0", "Accept": "application/json", } REQUEST_TIMEOUT = 20 REQUEST_RETRIES = 3 REQUEST_RETRY_DELAY = 1.25 CACHE_DIR = "/data/live_trading_cache" if os.path.exists("/data") else "./live_trading_cache" os.makedirs(CACHE_DIR, exist_ok=True) MAX_TRADING_ROWS = 6 DEFAULT_CLOCK = 60.0 MARKET_FILTER_CHOICES = [ "All Markets", "Passing Yards", "Passing TDs", "Rushing Yards", "Receiving Yards", "Anytime TD", ] # --------------------------------------------------------- # Market configuration # --------------------------------------------------------- @dataclass class MarketConfig: key: str label: str family: str distribution_type: str subcategory_id: Optional[str] live_stat_key: str live_stat_group: str parser_hint_keywords: List[str] = field(default_factory=list) MARKET_CONFIGS: Dict[str, MarketConfig] = { "passing_yards": MarketConfig( key="passing_yards", label="Passing Yards", family="yardage", distribution_type="yardage_normalish", subcategory_id=SUBCATEGORY_IDS.get("passing_yards"), live_stat_key="pass_yards", live_stat_group="pass", parser_hint_keywords=["passing yards", "pass yds", "pass yards"], ), "passing_tds": MarketConfig( key="passing_tds", label="Passing TDs", family="td_count", distribution_type="poisson_count", subcategory_id=SUBCATEGORY_IDS.get("passing_tds"), live_stat_key="pass_tds", live_stat_group="pass", parser_hint_keywords=["passing touchdowns", "pass tds", "pass touchdowns"], ), "rushing_yards": MarketConfig( key="rushing_yards", label="Rushing Yards", family="yardage", distribution_type="yardage_normalish", subcategory_id=SUBCATEGORY_IDS.get("rushing_yards"), live_stat_key="rush_yards", live_stat_group="rush", parser_hint_keywords=["rushing yards", "rush yds", "rush yards"], ), "receiving_yards": MarketConfig( key="receiving_yards", label="Receiving Yards", family="yardage", distribution_type="yardage_normalish", subcategory_id=SUBCATEGORY_IDS.get("receiving_yards"), live_stat_key="rec_yards", live_stat_group="rec", parser_hint_keywords=["receiving yards", "rec yds", "rec yards"], ), "anytime_td": MarketConfig( key="anytime_td", label="Anytime TD", family="td_count", distribution_type="poisson_count", subcategory_id=SUBCATEGORY_IDS.get("anytime_td"), live_stat_key="tds", live_stat_group="td", parser_hint_keywords=["anytime td", "any time td", "to score a touchdown"], ), } # Yardage markets need a target line to produce an actionable probability or fair price. # If an exact prematch line is unavailable from the current payload, the app uses a # defensible fallback target derived from the mean itself and clearly labels the row. YARDAGE_TARGET_RATIO = 1.00 YARDAGE_SD_FLOOR = 8.0 YARDAGE_SD_FACTOR = 0.35 VERY_LONGSHOT_DECIMAL_CUTOFF = 49.0 VERY_LONGSHOT_AMERICAN_CUTOFF = 5000 # ========================================================= # SESSION / STATE MODELS # ========================================================= @dataclass class EventSnapshot: event_id: str event_name: str home_team: str away_team: str event_status: str = "" start_time_utc: str = "" raw_metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class PlayerMarketSnapshot: available: bool market_key: str market_label: str decimal_odds: Optional[float] = None prematch_mean: Optional[float] = None target_line: Optional[float] = None source: str = "" notes: str = "" @dataclass class PlayerSnapshot: player_name: str markets: Dict[str, PlayerMarketSnapshot] = field(default_factory=dict) @dataclass class EventPlayerBoard: event: EventSnapshot players: Dict[str, PlayerSnapshot] = field(default_factory=dict) @dataclass class SessionSlate: session_date: str built_at_utc: str event_name_to_id: Dict[str, str] events_by_id: Dict[str, Dict[str, Any]] parser_status: Dict[str, Any] version: str = "2.0" def utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat() def today_key() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%d") def cache_path_for_today() -> str: return os.path.join(CACHE_DIR, f"{today_key()}_session_slate.json") # ========================================================= # NETWORK HELPERS # ========================================================= def fetch_json(url: str, retries: int = REQUEST_RETRIES, delay: float = REQUEST_RETRY_DELAY) -> dict: last_error = None for attempt in range(retries): try: response = requests.get(url, headers=HEADERS, timeout=REQUEST_TIMEOUT) response.raise_for_status() return response.json() except Exception as exc: last_error = exc if attempt < retries - 1: time.sleep(delay) raise last_error def build_event_subcategory_url(event_id: str, subcategory_id: str) -> str: markets_query = ( f"$filter=eventId eq '{event_id}' " f"AND clientMetadata/subCategoryId eq '{subcategory_id}' " f"AND tags/all(t: t ne 'SportcastBetBuilder')" ) return ( "https://sportsbook-nash.draftkings.com/sites/US-VA-SB/api/" "sportscontent/controldata/event/eventSubcategory/v1/markets" f"?isBatchable=false" f"&templateVars={event_id}%2C{subcategory_id}" f"&marketsQuery={quote(markets_query, safe='')}" "&entity=markets" ) # ========================================================= # CACHE HELPERS # ========================================================= def load_cached_slate() -> Optional[dict]: path = cache_path_for_today() if not os.path.exists(path): return None with open(path, "r", encoding="utf-8") as f: return json.load(f) def save_cached_slate(data: dict) -> None: with open(cache_path_for_today(), "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) # ========================================================= # GENERIC DATA HELPERS # ========================================================= def safe_float(value: Any) -> Optional[float]: try: if value is None or value == "": return None return float(value) except Exception: return None def safe_int(value: Any, default: int = 0) -> int: try: if value is None or value == "": return default return int(float(value)) except Exception: return default def clamp(value: float, lo: float, hi: float) -> float: return max(lo, min(hi, value)) def clean_name(value: Any) -> str: if value is None: return "" return str(value).strip() def split_event_name(event_name: str) -> Tuple[str, str]: """ Very light parsing for display. This is only cosmetic. """ name = clean_name(event_name) if " @ " in name: away, home = name.split(" @ ", 1) return home.strip(), away.strip() if " vs " in name.lower(): parts = name.split(" vs ") if len(parts) == 2: return parts[1].strip(), parts[0].strip() return "", "" def default_live_stats() -> Dict[str, Any]: return { "pass_yards": 0, "pass_tds": 0, "rush_yards": 0, "rec_yards": 0, "tds": 0, } def default_trader_row() -> Dict[str, Any]: return { "player": None, "pass_yards": 0, "pass_tds": 0, "rush_yards": 0, "rec_yards": 0, "tds": 0, } def fresh_trader_rows() -> List[Dict[str, Any]]: return [default_trader_row() for _ in range(MAX_TRADING_ROWS)] # ========================================================= # CLOCK HELPERS # ========================================================= def format_football_clock(clock_value: float) -> str: """ Converts a 60 -> 0 style game clock into a football-style quarter clock. Examples: 60.0 -> Q1 15:00 44.5 -> Q2 14:30 30.0 -> Q3 15:00 8.75 -> Q4 08:45 """ try: g = clamp(float(clock_value), 0.0, 60.0) except Exception: g = DEFAULT_CLOCK if g > 45: quarter = "Q1" quarter_remaining = 15 - (60 - g) elif g > 30: quarter = "Q2" quarter_remaining = 15 - (45 - g) elif g > 15: quarter = "Q3" quarter_remaining = 15 - (30 - g) else: quarter = "Q4" quarter_remaining = g # Convert fractional minutes to MM:SS whole_minutes = int(math.floor(quarter_remaining)) seconds = int(round((quarter_remaining - whole_minutes) * 60)) if seconds == 60: whole_minutes += 1 seconds = 0 whole_minutes = max(0, min(15, whole_minutes)) seconds = max(0, min(59, seconds)) return f"{quarter} {whole_minutes:02d}:{seconds:02d}" def format_clock_badge(clock_value: float) -> str: try: g = clamp(float(clock_value), 0.0, 60.0) except Exception: g = DEFAULT_CLOCK return f"{g:.2f} min" def elapsed_fraction(clock_value: float) -> float: try: g = clamp(float(clock_value), 0.0, 60.0) except Exception: g = DEFAULT_CLOCK return (60.0 - g) / 60.0 def remaining_fraction(clock_value: float) -> float: try: g = clamp(float(clock_value), 0.0, 60.0) except Exception: g = DEFAULT_CLOCK return g / 60.0 # ========================================================= # MATH / PRICING HELPERS # ========================================================= def implied_prob_from_decimal(decimal_odds: Any) -> Optional[float]: dec = safe_float(decimal_odds) if dec is None or dec <= 1: return None return 1.0 / dec def pm_mean_from_decimal(decimal_odds: Any) -> Optional[float]: """ Converts an anytime-style 1+ event probability into a Poisson mean. """ p = implied_prob_from_decimal(decimal_odds) if p is None or p <= 0 or p >= 1: return None return -math.log(1 - p) def american_from_probability(probability: Optional[float]) -> str: if probability is None or probability <= 0 or probability >= 1: return "" decimal_odds = 1.0 / probability if decimal_odds > VERY_LONGSHOT_DECIMAL_CUTOFF: return "DontOffer" if probability >= 0.5: american = -100 * probability / (1 - probability) else: american = 100 * (1 - probability) / probability american = int(round(american)) if american > VERY_LONGSHOT_AMERICAN_CUTOFF: return "DontOffer" return f"+{american}" if american > 0 else str(american) def decimal_from_probability(probability: Optional[float]) -> Optional[float]: if probability is None or probability <= 0 or probability >= 1: return None return 1.0 / probability def standard_normal_cdf(x: float) -> float: """ Error-function based standard normal CDF without SciPy dependency. """ return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) def poisson_prob_at_least(k: int, lmbda: Optional[float]) -> Optional[float]: if lmbda is None or lmbda < 0 or k < 0: return None cumulative = 0.0 for i in range(k): cumulative += math.exp(-lmbda) * (lmbda ** i) / math.factorial(i) return max(0.0, min(1.0, 1.0 - cumulative)) def time_adjusted_mean(clock_value: float, prematch_mean: Optional[float]) -> Optional[float]: """ Legacy time scaling retained and generalized for live-use continuity. This gives a remaining-game expectation proxy from a full-game prematch mean. """ if prematch_mean is None: return None try: g = clamp(float(clock_value), 0.0, 60.0) l = float(prematch_mean) except Exception: return None if g >= 45: return (g - 45) * (0.203 / 15.0) * l + (l * 0.7971) elif g >= 30: return (g - 30) * (0.312 / 15.0) * l + (l * 0.4851) elif g >= 15: return (g - 15) * (0.2073 / 15.0) * l + (l * 0.2778) elif g > 0: return g * (0.2778 / 15.0) * l else: return 0.0 def estimate_yardage_sd(prematch_mean: Optional[float]) -> Optional[float]: if prematch_mean is None: return None return max(YARDAGE_SD_FLOOR, abs(prematch_mean) * YARDAGE_SD_FACTOR) def time_adjusted_yardage_remaining_mean(clock_value: float, prematch_mean: Optional[float]) -> Optional[float]: if prematch_mean is None: return None return max(0.0, prematch_mean * remaining_fraction(clock_value)) def infer_yardage_target_line(prematch_mean: Optional[float]) -> Optional[float]: if prematch_mean is None: return None return round(prematch_mean * YARDAGE_TARGET_RATIO, 1) def pm_mean_from_yardage_line(line: Optional[float], lean_factor: float = 1.02) -> Optional[float]: """ Practical fallback for yardage markets when only a line is known or inferable. """ if line is None: return None return max(0.0, float(line) * lean_factor) def price_anytime_td(clock_value: float, prematch_mean: Optional[float], actual_tds: int) -> Dict[str, Any]: remaining_mean = time_adjusted_mean(clock_value, prematch_mean) if remaining_mean is None: return { "prematch_mean": None, "live_mean": None, "current_stat": actual_tds, "fair_probability": None, "fair_decimal": None, "fair_american": "", "notes": "Missing prematch mean", } if actual_tds >= 1: return { "prematch_mean": prematch_mean, "live_mean": 0.0, "current_stat": actual_tds, "fair_probability": 0.0, "fair_decimal": None, "fair_american": "", "notes": "Already scored; 1+ market settled", } fair_probability = poisson_prob_at_least(1, remaining_mean) return { "prematch_mean": prematch_mean, "live_mean": remaining_mean, "current_stat": actual_tds, "fair_probability": fair_probability, "fair_decimal": decimal_from_probability(fair_probability), "fair_american": american_from_probability(fair_probability), "notes": "Poisson 1+ from remaining-game mean", } def price_passing_tds(clock_value: float, prematch_mean: Optional[float], actual_pass_tds: int) -> Dict[str, Any]: remaining_mean = time_adjusted_mean(clock_value, prematch_mean) if remaining_mean is None: return { "prematch_mean": None, "live_mean": None, "current_stat": actual_pass_tds, "fair_probability": None, "fair_decimal": None, "fair_american": "", "notes": "Missing prematch mean", } fair_probability = poisson_prob_at_least(1, remaining_mean) notes = "Poisson price for next passing TD / 1+ remaining" if actual_pass_tds > 0: notes = f"Already has {actual_pass_tds} passing TD(s); pricing 1+ additional" return { "prematch_mean": prematch_mean, "live_mean": remaining_mean, "current_stat": actual_pass_tds, "fair_probability": fair_probability, "fair_decimal": decimal_from_probability(fair_probability), "fair_american": american_from_probability(fair_probability), "notes": notes, } def price_yardage_market( clock_value: float, prematch_mean: Optional[float], current_stat: float, target_line: Optional[float], market_label: str, ) -> Dict[str, Any]: """ Produces a fair price for the player to finish OVER a target line. The target can be extracted directly later from prematch market data. For now, the framework supports inferred targets when needed. """ if prematch_mean is None: return { "prematch_mean": None, "live_mean": None, "current_stat": current_stat, "fair_probability": None, "fair_decimal": None, "fair_american": "", "notes": "Missing prematch mean", } target_line = target_line if target_line is not None else infer_yardage_target_line(prematch_mean) remaining_mean = time_adjusted_yardage_remaining_mean(clock_value, prematch_mean) sd = estimate_yardage_sd(prematch_mean) if target_line is None or remaining_mean is None or sd is None or sd <= 0: return { "prematch_mean": prematch_mean, "live_mean": None, "current_stat": current_stat, "fair_probability": None, "fair_decimal": None, "fair_american": "", "notes": f"Could not derive {market_label} target/variance", } live_finish_mean = float(current_stat) + remaining_mean shortfall = target_line - float(current_stat) z = shortfall / sd fair_probability = 1.0 - standard_normal_cdf(z) note = f"Finish-over {target_line:.1f} via normal-style yardage proxy" if current_stat >= target_line: note = f"Already over inferred line {target_line:.1f}; pricing finish-over" return { "prematch_mean": prematch_mean, "live_mean": live_finish_mean, "current_stat": current_stat, "fair_probability": max(0.0, min(1.0, fair_probability)), "fair_decimal": decimal_from_probability(fair_probability), "fair_american": american_from_probability(fair_probability), "notes": note, "target_line": target_line, } def price_market( market_key: str, clock_value: float, market_snapshot: PlayerMarketSnapshot, live_stats: Dict[str, Any], ) -> Dict[str, Any]: cfg = MARKET_CONFIGS[market_key] if market_key == "anytime_td": return price_anytime_td( clock_value=clock_value, prematch_mean=market_snapshot.prematch_mean, actual_tds=safe_int(live_stats.get("tds"), 0), ) if market_key == "passing_tds": return price_passing_tds( clock_value=clock_value, prematch_mean=market_snapshot.prematch_mean, actual_pass_tds=safe_int(live_stats.get("pass_tds"), 0), ) if market_key == "passing_yards": return price_yardage_market( clock_value=clock_value, prematch_mean=market_snapshot.prematch_mean, current_stat=float(safe_float(live_stats.get("pass_yards")) or 0.0), target_line=market_snapshot.target_line, market_label=cfg.label, ) if market_key == "rushing_yards": return price_yardage_market( clock_value=clock_value, prematch_mean=market_snapshot.prematch_mean, current_stat=float(safe_float(live_stats.get("rush_yards")) or 0.0), target_line=market_snapshot.target_line, market_label=cfg.label, ) if market_key == "receiving_yards": return price_yardage_market( clock_value=clock_value, prematch_mean=market_snapshot.prematch_mean, current_stat=float(safe_float(live_stats.get("rec_yards")) or 0.0), target_line=market_snapshot.target_line, market_label=cfg.label, ) return { "prematch_mean": None, "live_mean": None, "current_stat": "", "fair_probability": None, "fair_decimal": None, "fair_american": "", "notes": "Unsupported market", } # ========================================================= # PARSERS # ========================================================= def parse_events(events_json: dict) -> Dict[str, Dict[str, Any]]: """ Returns a mapping keyed by event_id containing normalized event metadata. """ results: Dict[str, Dict[str, Any]] = {} for event in events_json.get("events", []) or []: event_id = clean_name(event.get("id")) if not event_id: continue event_name = ( clean_name(event.get("name")) or clean_name(event.get("eventName")) or clean_name(event.get("description")) or f"Event {event_id}" ) home_team, away_team = split_event_name(event_name) results[event_id] = { "event_id": event_id, "event_name": event_name, "home_team": clean_name(event.get("homeTeamName")) or home_team, "away_team": clean_name(event.get("awayTeamName")) or away_team, "event_status": clean_name(event.get("eventStatus") or event.get("status")), "start_time_utc": clean_name(event.get("startDate") or event.get("startTime")), "raw_metadata": { "groupId": event.get("groupId"), "leagueId": event.get("leagueId"), "id": event.get("id"), }, } return results def get_selection_price_decimal(selection: dict) -> Optional[float]: """ Attempts a few common fields found in different payload shapes. """ candidates = [ selection.get("oddsDecimal"), selection.get("trueOdds"), selection.get("price"), selection.get("displayOdds", {}).get("decimal"), selection.get("odds", {}).get("decimal"), ] for candidate in candidates: value = safe_float(candidate) if value is not None and value > 1: return value return None def get_selection_label(selection: dict) -> str: return ( clean_name(selection.get("label")) or clean_name(selection.get("name")) or clean_name(selection.get("description")) or clean_name(selection.get("runnerName")) or clean_name(selection.get("outcomeLabel")) ) def get_market_name(market: dict) -> str: return ( clean_name(market.get("name")) or clean_name(market.get("description")) or clean_name(market.get("marketName")) or clean_name(market.get("subcategoryName")) ) def infer_market_target_line(market: dict, selection: dict) -> Optional[float]: """ Best-effort extraction for line/points style props. """ candidates = [ market.get("line"), market.get("points"), market.get("value"), selection.get("line"), selection.get("points"), selection.get("value"), selection.get("participantLine"), ] for candidate in candidates: value = safe_float(candidate) if value is not None: return value return None def infer_player_market_mean( market_key: str, market_name: str, selection: dict, decimal_odds: Optional[float], target_line: Optional[float], ) -> Tuple[Optional[float], str]: """ Centralized market-to-mean translation. """ if market_key in {"anytime_td", "passing_tds"}: mean = pm_mean_from_decimal(decimal_odds) return mean, "implied_from_decimal" if market_key in {"passing_yards", "rushing_yards", "receiving_yards"}: line = target_line if line is None: line = infer_yardage_target_line(pm_mean_from_decimal(decimal_odds) or None) mean = pm_mean_from_yardage_line(line) if line is not None else None return mean, "derived_from_line" return None, "unknown" def should_accept_selection_for_market(market_key: str, market_name: str, selection_label: str) -> bool: """ Flexible acceptance logic. The current feed structure may vary over time, so this intentionally stays permissive while avoiding obvious blanks. """ if not selection_label: return False market_name_l = market_name.lower() label_l = selection_label.lower() cfg = MARKET_CONFIGS[market_key] if market_key == "anytime_td": return True keyword_hits = any(keyword in market_name_l for keyword in cfg.parser_hint_keywords) if keyword_hits: return True # Soft fallback: if the selection looks like a player and we have a price. # This keeps the parser resilient when payload naming changes. if len(selection_label.split()) >= 2 and not label_l.startswith("over") and not label_l.startswith("under"): return True return False def extract_market_snapshots_from_payload( payload: dict, market_key: str, ) -> Dict[str, PlayerMarketSnapshot]: """ Generic extraction engine for a single event+market payload. """ results: Dict[str, PlayerMarketSnapshot] = {} markets = payload.get("markets", []) or [] selections = payload.get("selections", []) or payload.get("outcomes", []) or [] selections_by_market: Dict[Any, List[dict]] = {} for selection in selections: market_id = selection.get("marketId") or selection.get("market_id") if market_id is None: continue selections_by_market.setdefault(market_id, []).append(selection) for market in markets: market_id = market.get("id") market_name = get_market_name(market) if market_id is None: continue for selection in selections_by_market.get(market_id, []): player_name = get_selection_label(selection) if not should_accept_selection_for_market(market_key, market_name, player_name): continue decimal_odds = get_selection_price_decimal(selection) target_line = infer_market_target_line(market, selection) prematch_mean, mean_source = infer_player_market_mean( market_key=market_key, market_name=market_name, selection=selection, decimal_odds=decimal_odds, target_line=target_line, ) notes = [] if decimal_odds is None: notes.append("missing_decimal_odds") if prematch_mean is None: notes.append("missing_prematch_mean") if target_line is None and market_key in {"passing_yards", "rushing_yards", "receiving_yards"}: notes.append("target_line_inferred_later") results[player_name] = PlayerMarketSnapshot( available=True, market_key=market_key, market_label=MARKET_CONFIGS[market_key].label, decimal_odds=decimal_odds, prematch_mean=prematch_mean, target_line=target_line, source=mean_source, notes="; ".join(notes), ) return results # ========================================================= # SESSION SLATE BUILD # ========================================================= def merge_market_snapshot_into_event_board( board: EventPlayerBoard, market_key: str, market_results: Dict[str, PlayerMarketSnapshot], ) -> None: for player_name, market_snapshot in market_results.items(): if player_name not in board.players: board.players[player_name] = PlayerSnapshot(player_name=player_name, markets={}) board.players[player_name].markets[market_key] = market_snapshot def build_empty_market_placeholders(board: EventPlayerBoard) -> None: """ Ensures every player has every supported market key, even if unavailable. That keeps downstream pricing/table logic simple and stable. """ for player in board.players.values(): for market_key, cfg in MARKET_CONFIGS.items(): if market_key not in player.markets: player.markets[market_key] = PlayerMarketSnapshot( available=False, market_key=market_key, market_label=cfg.label, decimal_odds=None, prematch_mean=None, target_line=None, source="not_loaded", notes="Market not found in frozen prematch snapshot", ) def build_daily_slate() -> dict: events_json = fetch_json(EVENTS_API_URL) parsed_events = parse_events(events_json) event_name_to_id: Dict[str, str] = {} events_by_id: Dict[str, Dict[str, Any]] = {} parser_status: Dict[str, Any] = { "successful_events": 0, "failed_events": 0, "market_fetches": [], } for event_id, event_meta in parsed_events.items(): event_snapshot = EventSnapshot(**event_meta) event_name_to_id[event_snapshot.event_name] = event_id event_board = EventPlayerBoard(event=event_snapshot, players={}) # Attempt each configured market. Failures are captured but do not stop the build. for market_key, cfg in MARKET_CONFIGS.items(): if not cfg.subcategory_id: parser_status["market_fetches"].append({ "event_id": event_id, "market_key": market_key, "status": "skipped_no_subcategory_id", }) continue try: payload = fetch_json(build_event_subcategory_url(event_id, cfg.subcategory_id)) market_results = extract_market_snapshots_from_payload(payload, market_key) merge_market_snapshot_into_event_board(event_board, market_key, market_results) parser_status["market_fetches"].append({ "event_id": event_id, "market_key": market_key, "status": "ok", "players_found": len(market_results), }) except Exception as exc: parser_status["market_fetches"].append({ "event_id": event_id, "market_key": market_key, "status": "error", "error": str(exc), }) build_empty_market_placeholders(event_board) if event_board.players: parser_status["successful_events"] += 1 else: parser_status["failed_events"] += 1 events_by_id[event_id] = { "event": asdict(event_board.event), "players": { player_name: { "player_name": player_snapshot.player_name, "markets": { m_key: asdict(m_snapshot) for m_key, m_snapshot in player_snapshot.markets.items() }, } for player_name, player_snapshot in event_board.players.items() }, } slate = SessionSlate( session_date=today_key(), built_at_utc=utc_now_iso(), event_name_to_id=event_name_to_id, events_by_id=events_by_id, parser_status=parser_status, version="2.0", ) slate_dict = asdict(slate) save_cached_slate(slate_dict) return slate_dict def ensure_daily_slate() -> Tuple[dict, str]: cached = load_cached_slate() if cached: return cached, f"Loaded cached session slate for {cached.get('session_date', today_key())}" built = build_daily_slate() return built, f"Built new session slate for {built.get('session_date', today_key())}" # ========================================================= # SESSION ACCESS HELPERS # ========================================================= def blank_output_df() -> pd.DataFrame: return pd.DataFrame(columns=[ "Player", "Market", "Prematch Mean", "Live Mean", "Current Stat", "Fair Price", "Fair Prob", "Notes / Status", ]) def get_event_names_from_slate(session_slate: dict) -> List[str]: if not isinstance(session_slate, dict): return [] return sorted(list((session_slate.get("event_name_to_id") or {}).keys())) def get_event_id_from_name(session_slate: dict, event_name: str) -> Optional[str]: if not isinstance(session_slate, dict): return None return (session_slate.get("event_name_to_id") or {}).get(event_name) def get_event_board(session_slate: dict, event_name: str) -> Optional[dict]: event_id = get_event_id_from_name(session_slate, event_name) if not event_id: return None return (session_slate.get("events_by_id") or {}).get(event_id) def get_player_choices_for_event(session_slate: dict, event_name: str) -> List[str]: board = get_event_board(session_slate, event_name) if not board: return [] players = board.get("players") or {} return sorted(players.keys()) def get_event_summary_markdown(session_slate: dict, event_name: Optional[str]) -> str: if not session_slate: return _info_card("No frozen session slate loaded.") if not event_name: event_count = len(get_event_names_from_slate(session_slate)) built_at = session_slate.get("built_at_utc", "") return _info_card( f"Frozen slate active • {event_count} games • Built {built_at}
" f"Select a game to load trader rows and frozen prematch player markets." ) board = get_event_board(session_slate, event_name) if not board: return _warning_card("Selected event not found in frozen session slate.") event = board.get("event") or {} player_count = len(board.get("players") or {}) home_team = clean_name(event.get("home_team")) away_team = clean_name(event.get("away_team")) event_status = clean_name(event.get("event_status")) or "Prematch / Snapshot" matchup = event_name if not (away_team and home_team) else f"{away_team} @ {home_team}" return _info_card( f"{matchup}
" f"Players in snapshot: {player_count} • Status: {event_status}
" f"All pricing below is driven from the frozen session slate for this session." ) # ========================================================= # OUTPUT BUILDERS # ========================================================= def format_float_for_table(value: Optional[float], digits: int = 2) -> str: if value is None: return "" try: return f"{float(value):.{digits}f}" except Exception: return "" def market_filter_to_keys(selected_filter: str) -> List[str]: if selected_filter == "All Markets": return list(MARKET_CONFIGS.keys()) mapping = {cfg.label: key for key, cfg in MARKET_CONFIGS.items()} key = mapping.get(selected_filter) return [key] if key else list(MARKET_CONFIGS.keys()) def build_market_row( player_name: str, market_key: str, market_snapshot: PlayerMarketSnapshot, pricing_result: Dict[str, Any], ) -> Dict[str, Any]: fair_prob = pricing_result.get("fair_probability") fair_prob_str = f"{fair_prob:.3%}" if fair_prob is not None else "" notes = pricing_result.get("notes", "") if market_snapshot.notes: notes = f"{notes} | {market_snapshot.notes}" if notes else market_snapshot.notes return { "Player": player_name, "Market": market_snapshot.market_label, "Prematch Mean": format_float_for_table(pricing_result.get("prematch_mean"), 3), "Live Mean": format_float_for_table(pricing_result.get("live_mean"), 3), "Current Stat": pricing_result.get("current_stat", ""), "Fair Price": pricing_result.get("fair_american", ""), "Fair Prob": fair_prob_str, "Notes / Status": notes, } def build_output_table( clock_value: float, market_filter: str, event_name: str, session_slate: dict, trader_rows_state: List[Dict[str, Any]], ) -> Tuple[pd.DataFrame, str]: if not session_slate: return blank_output_df(), "Session slate not initialized. Click Initialize Session Slate first." if not event_name: return blank_output_df(), "Select a game from the frozen session slate." board = get_event_board(session_slate, event_name) if not board: return blank_output_df(), "Selected game is missing from the current session slate." players_map = board.get("players") or {} selected_market_keys = market_filter_to_keys(market_filter) output_rows: List[Dict[str, Any]] = [] for row in trader_rows_state or []: player_name = clean_name(row.get("player")) if not player_name: continue player_blob = players_map.get(player_name) if not player_blob: output_rows.append({ "Player": player_name, "Market": "", "Prematch Mean": "", "Live Mean": "", "Current Stat": "", "Fair Price": "", "Fair Prob": "", "Notes / Status": "Player missing from frozen session snapshot for selected event", }) continue player_markets = player_blob.get("markets") or {} live_stats = { "pass_yards": safe_float(row.get("pass_yards")) or 0.0, "pass_tds": safe_int(row.get("pass_tds"), 0), "rush_yards": safe_float(row.get("rush_yards")) or 0.0, "rec_yards": safe_float(row.get("rec_yards")) or 0.0, "tds": safe_int(row.get("tds"), 0), } for market_key in selected_market_keys: cfg = MARKET_CONFIGS[market_key] market_blob = player_markets.get(market_key) if not market_blob: output_rows.append({ "Player": player_name, "Market": cfg.label, "Prematch Mean": "", "Live Mean": "", "Current Stat": "", "Fair Price": "", "Fair Prob": "", "Notes / Status": "Market missing from frozen prematch board", }) continue snapshot = PlayerMarketSnapshot(**market_blob) if not snapshot.available: output_rows.append({ "Player": player_name, "Market": cfg.label, "Prematch Mean": "", "Live Mean": "", "Current Stat": live_stats.get(cfg.live_stat_key, ""), "Fair Price": "", "Fair Prob": "", "Notes / Status": snapshot.notes or "Market unavailable in snapshot", }) continue pricing_result = price_market( market_key=market_key, clock_value=clock_value, market_snapshot=snapshot, live_stats=live_stats, ) output_rows.append(build_market_row(player_name, market_key, snapshot, pricing_result)) if not output_rows: return blank_output_df(), "No active player rows selected." df = pd.DataFrame(output_rows) return df, f"Updated {len(df)} market rows from frozen session slate." # ========================================================= # LIVE STATS / FUTURE AUTO UPDATE HOOKS # ========================================================= def get_live_game_state_stub(event_id: str) -> Dict[str, Any]: """ Intentional extension point. In a future build this function can: - call a live UFL/NFL boxscore endpoint - poll a game-state feed - map live player stats into the trader row schema - feed auto-refresh / timer-based callbacks The rest of the application already separates: frozen prematch session slate vs. live game state inputs. """ return { "event_id": event_id, "clock": None, "players": {}, "source": "stub", } # ========================================================= # STATUS / DISPLAY HELPERS # ========================================================= def _status_chip(label: str, tone: str = "neutral") -> str: return f"{label}" def _info_card(text: str) -> str: return f"
{text}
" def _warning_card(text: str) -> str: return f"
{text}
" def _success_card(text: str) -> str: return f"
{text}
" def build_session_status_html(session_slate: dict) -> str: if not session_slate: return ( "
" f"{_status_chip('No Session Slate', 'danger')}" "Click Initialize Session Slate to freeze the day’s prematch board." "
" ) event_count = len(get_event_names_from_slate(session_slate)) built_at = clean_name(session_slate.get("built_at_utc")) version = clean_name(session_slate.get("version") or "2.0") return ( "
" f"{_status_chip('Frozen Session Slate Active', 'success')}" f"{_status_chip(f'{event_count} Games', 'neutral')}" f"{_status_chip(f'v{version}', 'accent')}" f"Built at {built_at}" "
" ) def build_clock_panel_html(clock_value: float) -> str: return ( "
" "
Shared Game Clock
" f"
{format_football_clock(clock_value)}
" f"
{format_clock_badge(clock_value)} remaining
" "
All player markets in this session use the same live game clock.
" "
" ) def build_footer_note_html() -> str: return _info_card( "The app separates frozen prematch session data from live trading inputs. " "That makes it robust when prematch links disappear and ready for future live endpoint polling." ) # ========================================================= # UI ACTIONS # ========================================================= def initialize_app(): empty_players = [gr.update(choices=[], value=None) for _ in range(MAX_TRADING_ROWS)] empty_stats = [0 for _ in range(MAX_TRADING_ROWS * 5)] return ( gr.update(choices=[], value=None), {}, fresh_trader_rows(), build_session_status_html({}), build_clock_panel_html(DEFAULT_CLOCK), _info_card("Page loaded. Initialize the session slate to freeze today’s offered prematch board."), _info_card("No game selected yet."), blank_output_df(), *empty_players, *empty_stats, ) def initialize_session_slate(): try: session_slate, status = ensure_daily_slate() event_names = get_event_names_from_slate(session_slate) empty_players = [gr.update(choices=[], value=None) for _ in range(MAX_TRADING_ROWS)] empty_stats = [0 for _ in range(MAX_TRADING_ROWS * 5)] return ( gr.update(choices=event_names, value=None), session_slate, fresh_trader_rows(), build_session_status_html(session_slate), build_clock_panel_html(DEFAULT_CLOCK), _success_card( f"{status}. Session snapshot initialized with {len(event_names)} games. " "This page will now use the frozen session slate for the entire session." ), get_event_summary_markdown(session_slate, None), blank_output_df(), *empty_players, *empty_stats, ) except Exception as exc: empty_players = [gr.update(choices=[], value=None) for _ in range(MAX_TRADING_ROWS)] empty_stats = [0 for _ in range(MAX_TRADING_ROWS * 5)] return ( gr.update(choices=[], value=None), {}, fresh_trader_rows(), build_session_status_html({}), build_clock_panel_html(DEFAULT_CLOCK), _warning_card(f"Session initialization failed: {exc}"), _info_card("No game selected."), blank_output_df(), *empty_players, *empty_stats, ) def rebuild_session_slate(): try: session_slate = build_daily_slate() event_names = get_event_names_from_slate(session_slate) empty_players = [gr.update(choices=[], value=None) for _ in range(MAX_TRADING_ROWS)] empty_stats = [0 for _ in range(MAX_TRADING_ROWS * 5)] return ( gr.update(choices=event_names, value=None), session_slate, fresh_trader_rows(), build_session_status_html(session_slate), build_clock_panel_html(DEFAULT_CLOCK), _success_card( f"Rebuilt frozen session slate from live source with {len(event_names)} games. " "This session is now using the refreshed snapshot." ), get_event_summary_markdown(session_slate, None), blank_output_df(), *empty_players, *empty_stats, ) except Exception as exc: return ( gr.update(), gr.skip(), fresh_trader_rows(), gr.skip(), build_clock_panel_html(DEFAULT_CLOCK), _warning_card(f"Session rebuild failed: {exc}"), gr.skip(), blank_output_df(), *[gr.update() for _ in range(MAX_TRADING_ROWS)], *[0 for _ in range(MAX_TRADING_ROWS * 5)], ) def load_players_for_event(event_name: str, session_slate: dict): if not session_slate: empty_players = [gr.update(choices=[], value=None) for _ in range(MAX_TRADING_ROWS)] empty_stats = [0 for _ in range(MAX_TRADING_ROWS * 5)] return ( fresh_trader_rows(), _warning_card("Session slate not initialized. Click Initialize Session Slate first."), _info_card("No active event."), blank_output_df(), *empty_players, *empty_stats, ) player_choices = get_player_choices_for_event(session_slate, event_name) dropdown_updates = [gr.update(choices=player_choices, value=None) for _ in range(MAX_TRADING_ROWS)] zeroed_stats = [0 for _ in range(MAX_TRADING_ROWS * 5)] trader_rows = fresh_trader_rows() if not event_name: return ( trader_rows, _info_card("No game selected."), get_event_summary_markdown(session_slate, None), blank_output_df(), *dropdown_updates, *zeroed_stats, ) summary = get_event_summary_markdown(session_slate, event_name) status_html = _success_card( f"Loaded {len(player_choices)} frozen players for {event_name}. " "Trader rows are reset and ready for live stat input." ) return ( trader_rows, status_html, summary, blank_output_df(), *dropdown_updates, *zeroed_stats, ) def sync_trader_rows( p1, py1, ptd1, ry1, rec1, td1, p2, py2, ptd2, ry2, rec2, td2, p3, py3, ptd3, ry3, rec3, td3, p4, py4, ptd4, ry4, rec4, td4, p5, py5, ptd5, ry5, rec5, td5, p6, py6, ptd6, ry6, rec6, td6, ) -> List[Dict[str, Any]]: flat = [ (p1, py1, ptd1, ry1, rec1, td1), (p2, py2, ptd2, ry2, rec2, td2), (p3, py3, ptd3, ry3, rec3, td3), (p4, py4, ptd4, ry4, rec4, td4), (p5, py5, ptd5, ry5, rec5, td5), (p6, py6, ptd6, ry6, rec6, td6), ] rows = [] for player, pass_yards, pass_tds, rush_yards, rec_yards, tds in flat: rows.append({ "player": player, "pass_yards": safe_float(pass_yards) or 0.0, "pass_tds": safe_int(pass_tds, 0), "rush_yards": safe_float(rush_yards) or 0.0, "rec_yards": safe_float(rec_yards) or 0.0, "tds": safe_int(tds, 0), }) return rows def refresh_output( clock_value: float, market_filter: str, event_name: str, session_slate: dict, p1, py1, ptd1, ry1, rec1, td1, p2, py2, ptd2, ry2, rec2, td2, p3, py3, ptd3, ry3, rec3, td3, p4, py4, ptd4, ry4, rec4, td4, p5, py5, ptd5, ry5, rec5, td5, p6, py6, ptd6, ry6, rec6, td6, ): trader_rows = sync_trader_rows( p1, py1, ptd1, ry1, rec1, td1, p2, py2, ptd2, ry2, rec2, td2, p3, py3, ptd3, ry3, rec3, td3, p4, py4, ptd4, ry4, rec4, td4, p5, py5, ptd5, ry5, rec5, td5, p6, py6, ptd6, ry6, rec6, td6, ) output_df, status = build_output_table( clock_value=clock_value, market_filter=market_filter, event_name=event_name, session_slate=session_slate, trader_rows_state=trader_rows, ) return ( trader_rows, output_df, build_clock_panel_html(clock_value), _info_card(status), ) # ========================================================= # CSS # ========================================================= CUSTOM_CSS = """ :root { --bg: #0d1117; --panel: #121826; --panel-2: #161f31; --card: #0f1726; --border: #243146; --muted: #91a0b8; --text: #eef3ff; --text-soft: #c5d0e3; --accent: #31c48d; --accent-2: #4f8cff; --warning: #ffcc66; --danger: #ff7b7b; --success: #34d399; } html, body, .gradio-container { background: linear-gradient(180deg, #0b0f15 0%, #0d1117 100%) !important; color: var(--text) !important; font-family: Inter, ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Helvetica, Arial, sans-serif !important; } .gradio-container { max-width: 1500px !important; } h1, h2, h3, h4, p, label, span, div { color: var(--text) !important; } .app-shell { padding: 10px 0 24px 0; } .hero { background: radial-gradient(circle at top right, rgba(79, 140, 255, 0.18), transparent 35%), radial-gradient(circle at top left, rgba(49, 196, 141, 0.16), transparent 35%), linear-gradient(180deg, rgba(18, 24, 38, 0.98), rgba(12, 18, 29, 0.98)); border: 1px solid var(--border); border-radius: 22px; padding: 22px 24px 18px 24px; margin-bottom: 16px; box-shadow: 0 16px 40px rgba(0,0,0,0.28); } .hero-title { font-size: 30px; font-weight: 800; letter-spacing: -0.02em; margin-bottom: 6px; } .hero-subtitle { font-size: 14px; color: var(--text-soft) !important; line-height: 1.5; } .panel { background: linear-gradient(180deg, rgba(18, 24, 38, 0.98), rgba(12, 18, 29, 0.98)); border: 1px solid var(--border); border-radius: 20px; padding: 16px 16px 14px 16px; margin-bottom: 16px; box-shadow: 0 10px 28px rgba(0,0,0,0.22); } .panel-tight { padding: 14px; } .panel-title { font-size: 16px; font-weight: 800; letter-spacing: -0.01em; margin-bottom: 10px; } .panel-sub { color: var(--muted) !important; font-size: 13px; margin-top: -2px; margin-bottom: 10px; } .status-row { display: flex; gap: 8px; flex-wrap: wrap; align-items: center; } .status-detail { color: var(--text-soft) !important; font-size: 13px; margin-left: 4px; } .status-chip { display: inline-flex; align-items: center; gap: 6px; padding: 6px 10px; border-radius: 999px; font-size: 12px; font-weight: 800; border: 1px solid transparent; } .status-chip.success { background: rgba(52, 211, 153, 0.12); color: #8ef0c7 !important; border-color: rgba(52, 211, 153, 0.25); } .status-chip.danger { background: rgba(255, 123, 123, 0.12); color: #ffaaaa !important; border-color: rgba(255, 123, 123, 0.25); } .status-chip.neutral { background: rgba(145, 160, 184, 0.12); color: #d8e1ef !important; border-color: rgba(145, 160, 184, 0.18); } .status-chip.accent { background: rgba(79, 140, 255, 0.12); color: #a8c4ff !important; border-color: rgba(79, 140, 255, 0.22); } .inline-info-card, .inline-warning-card, .inline-success-card { border-radius: 14px; padding: 12px 14px; font-size: 13px; line-height: 1.45; border: 1px solid var(--border); } .inline-info-card { background: rgba(79, 140, 255, 0.07); color: var(--text-soft) !important; } .inline-warning-card { background: rgba(255, 204, 102, 0.08); color: #ffe2a8 !important; border-color: rgba(255, 204, 102, 0.22); } .inline-success-card { background: rgba(52, 211, 153, 0.08); color: #b2f2d9 !important; border-color: rgba(52, 211, 153, 0.22); } .clock-shell { background: radial-gradient(circle at top right, rgba(49, 196, 141, 0.18), transparent 35%), linear-gradient(180deg, rgba(18, 24, 38, 0.98), rgba(13, 23, 39, 0.98)); border: 1px solid rgba(49, 196, 141, 0.18); border-radius: 18px; padding: 16px 18px; min-height: 160px; display: flex; flex-direction: column; justify-content: center; } .clock-label { font-size: 12px; text-transform: uppercase; letter-spacing: 0.08em; color: #9de6c6 !important; font-weight: 700; margin-bottom: 8px; } .clock-main { font-size: 38px; font-weight: 900; letter-spacing: -0.03em; line-height: 1.0; margin-bottom: 8px; } .clock-sub { font-size: 16px; font-weight: 700; color: var(--text-soft) !important; margin-bottom: 6px; } .clock-foot { font-size: 12px; color: var(--muted) !important; } .trader-row-title { font-size: 13px; font-weight: 800; color: var(--text-soft) !important; margin-bottom: 8px; letter-spacing: 0.02em; } .footer-note { margin-top: 4px; } button { border-radius: 12px !important; border: 1px solid var(--border) !important; font-weight: 700 !important; } button.primary { box-shadow: 0 8px 22px rgba(49, 196, 141, 0.18); } .gradio-container .gr-button { min-height: 42px !important; } .gradio-container input, .gradio-container textarea, .gradio-container select { background: rgba(10, 15, 24, 0.95) !important; color: var(--text) !important; border: 1px solid var(--border) !important; border-radius: 12px !important; } .gradio-container .gr-form { border: none !important; } .gradio-container .gr-box, .gradio-container .gr-panel { background: transparent !important; } .gradio-container table { background: rgba(13,17,23,0.98) !important; border-collapse: collapse !important; color: var(--text) !important; } .gradio-container th { background: #172033 !important; color: var(--text) !important; font-weight: 800 !important; border: 1px solid var(--border) !important; padding: 10px !important; } .gradio-container td { background: rgba(10, 15, 24, 0.98) !important; color: var(--text-soft) !important; border: 1px solid var(--border) !important; padding: 9px !important; } .gradio-container .wrap.svelte-1ipelgc { border-radius: 16px !important; overflow: hidden !important; } hr { border-color: var(--border) !important; } .gr-markdown p { margin-bottom: 0.35rem !important; } """ # ========================================================= # UI # ========================================================= with gr.Blocks(css=CUSTOM_CSS, title=APP_TITLE, theme=gr.themes.Base()) as demo: session_slate_state = gr.State({}) trader_rows_state = gr.State(fresh_trader_rows()) with gr.Column(elem_classes=["app-shell"]): with gr.Group(elem_classes=["hero"]): gr.Markdown( f"""
{APP_TITLE}
{APP_SUBTITLE}
""" ) session_status_html = gr.HTML(build_session_status_html({})) with gr.Row(equal_height=True): with gr.Column(scale=8): with gr.Group(elem_classes=["panel"]): gr.Markdown("
Session Controls
") gr.Markdown( "
Freeze the prematch board for the current session. " "All later pricing uses the stored snapshot even if the source link disappears.
" ) with gr.Row(): init_session_btn = gr.Button("Initialize Session Slate", variant="primary") rebuild_session_btn = gr.Button("Rebuild Session Slate") status_html = gr.HTML(_info_card("Page loaded. Initialize the session slate to begin.")) with gr.Column(scale=4): with gr.Group(elem_classes=["panel", "panel-tight"]): gr.Markdown("
Clock
") clock_panel_html = gr.HTML(build_clock_panel_html(DEFAULT_CLOCK)) clock_input = gr.Number( label="Numeric Game Clock (60 → 0)", value=DEFAULT_CLOCK, precision=2, minimum=0, maximum=60, ) with gr.Row(equal_height=True): with gr.Column(scale=7): with gr.Group(elem_classes=["panel"]): gr.Markdown("
Game Selection
") gr.Markdown("
Load one frozen game from the active session slate.
") event_dropdown = gr.Dropdown( label="Select Game", choices=[], value=None, interactive=True, ) event_summary_html = gr.HTML(_info_card("No game selected.")) with gr.Column(scale=5): with gr.Group(elem_classes=["panel"]): gr.Markdown("
Market View
") gr.Markdown("
View all markets together or focus on one trading family.
") market_filter_dropdown = gr.Dropdown( label="Market Filter", choices=MARKET_FILTER_CHOICES, value="All Markets", interactive=True, ) footer_note_html = gr.HTML(build_footer_note_html(), elem_classes=["footer-note"]) with gr.Group(elem_classes=["panel"]): gr.Markdown("
Player Trading Rows
") gr.Markdown( "
Select players from the frozen session board and enter live stats. " "The data model is already structured for future auto-populated live endpoints.
" ) player_dropdowns = [] stat_inputs = [] for idx in range(MAX_TRADING_ROWS): gr.Markdown(f"
Trader Row {idx + 1}
") with gr.Row(): player_dd = gr.Dropdown( label="Player", choices=[], value=None, scale=3, ) pass_yards = gr.Number(label="Pass Yds", value=0, precision=1, scale=1) pass_tds = gr.Number(label="Pass TDs", value=0, precision=0, scale=1) rush_yards = gr.Number(label="Rush Yds", value=0, precision=1, scale=1) rec_yards = gr.Number(label="Rec Yds", value=0, precision=1, scale=1) tds = gr.Number(label="TDs", value=0, precision=0, scale=1) player_dropdowns.append(player_dd) stat_inputs.extend([pass_yards, pass_tds, rush_yards, rec_yards, tds]) with gr.Group(elem_classes=["panel"]): gr.Markdown("
Live Pricing Output
") gr.Markdown( "
Trader-facing output table showing live fair pricing across markets " "for the selected rows.
" ) output_table = gr.Dataframe( value=blank_output_df(), interactive=False, wrap=True, label="Market Output Table", ) # ----------------------------------------------------- # Initial page load # ----------------------------------------------------- demo.load( fn=initialize_app, outputs=[ event_dropdown, session_slate_state, trader_rows_state, session_status_html, clock_panel_html, status_html, event_summary_html, output_table, *player_dropdowns, *stat_inputs, ], ) # ----------------------------------------------------- # Session actions # ----------------------------------------------------- init_session_btn.click( fn=initialize_session_slate, outputs=[ event_dropdown, session_slate_state, trader_rows_state, session_status_html, clock_panel_html, status_html, event_summary_html, output_table, *player_dropdowns, *stat_inputs, ], ) rebuild_session_btn.click( fn=rebuild_session_slate, outputs=[ event_dropdown, session_slate_state, trader_rows_state, session_status_html, clock_panel_html, status_html, event_summary_html, output_table, *player_dropdowns, *stat_inputs, ], ) # ----------------------------------------------------- # Event change # ----------------------------------------------------- event_dropdown.change( fn=load_players_for_event, inputs=[event_dropdown, session_slate_state], outputs=[ trader_rows_state, status_html, event_summary_html, output_table, *player_dropdowns, *stat_inputs, ], ) # ----------------------------------------------------- # Shared refresh pipeline # ----------------------------------------------------- refresh_inputs = [ clock_input, market_filter_dropdown, event_dropdown, session_slate_state, player_dropdowns[0], stat_inputs[0], stat_inputs[1], stat_inputs[2], stat_inputs[3], stat_inputs[4], player_dropdowns[1], stat_inputs[5], stat_inputs[6], stat_inputs[7], stat_inputs[8], stat_inputs[9], player_dropdowns[2], stat_inputs[10], stat_inputs[11], stat_inputs[12], stat_inputs[13], stat_inputs[14], player_dropdowns[3], stat_inputs[15], stat_inputs[16], stat_inputs[17], stat_inputs[18], stat_inputs[19], player_dropdowns[4], stat_inputs[20], stat_inputs[21], stat_inputs[22], stat_inputs[23], stat_inputs[24], player_dropdowns[5], stat_inputs[25], stat_inputs[26], stat_inputs[27], stat_inputs[28], stat_inputs[29], ] refresh_outputs = [ trader_rows_state, output_table, clock_panel_html, status_html, ] refresh_components = [clock_input, market_filter_dropdown] + player_dropdowns + stat_inputs for component in refresh_components: component.change( fn=refresh_output, inputs=refresh_inputs, outputs=refresh_outputs, ) if __name__ == "__main__": demo.launch()