josh
e0db8f0859
Unify overlapping late-P3 bonuses into a single score-state lookup, add high-scoring and goal-spike signals, and tighten every component ceiling so filling the hype bar is reserved for genuinely rare moments. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
582 lines
21 KiB
Python
582 lines
21 KiB
Python
import logging
|
||
import sqlite3
|
||
from datetime import datetime, timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from app.config import DB_PATH
|
||
|
||
EASTERN = ZoneInfo("America/New_York")
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Maps (home_team_name, away_team_name) -> (home_score, away_score)
|
||
_score_cache: dict[tuple[str, str], tuple[int, int]] = {}
|
||
|
||
# Maps (home_team_name, away_team_name) -> max score differential seen
|
||
_comeback_tracker: dict[tuple[str, str], int] = {}
|
||
|
||
|
||
def format_record(record):
|
||
if record == "N/A":
|
||
return "N/A"
|
||
else:
|
||
parts = record.split("-")
|
||
formatted_parts = [part.zfill(2) for part in parts]
|
||
return "-".join(formatted_parts)
|
||
|
||
|
||
def parse_games(scoreboard_data):
|
||
if not scoreboard_data:
|
||
return []
|
||
|
||
extracted_info = []
|
||
for game in scoreboard_data.get("games", []):
|
||
game_state = convert_game_state(game["gameState"])
|
||
priority_comps = _priority_components(game)
|
||
momentum = _momentum_components(game)
|
||
importance_comps = _importance_components(game)
|
||
total_priority = (
|
||
priority_comps["total"] + momentum["total"] + importance_comps["total"]
|
||
)
|
||
extracted_info.append(
|
||
{
|
||
"Home Team": game["homeTeam"]["name"]["default"],
|
||
"Home Score": game["homeTeam"]["score"]
|
||
if game_state != "PRE"
|
||
else "N/A",
|
||
"Away Team": game["awayTeam"]["name"]["default"],
|
||
"Away Score": game["awayTeam"]["score"]
|
||
if game_state != "PRE"
|
||
else "N/A",
|
||
"Home Logo": game["homeTeam"]["logo"],
|
||
"Away Logo": game["awayTeam"]["logo"],
|
||
"Game State": game_state,
|
||
"Game Type": game.get("gameType", 2),
|
||
"Period": get_period(game),
|
||
"Time Remaining": get_time_remaining(game),
|
||
"Time Running": game["clock"]["running"]
|
||
if game_state == "LIVE"
|
||
else "N/A",
|
||
"Intermission": game["clock"]["inIntermission"]
|
||
if game_state == "LIVE"
|
||
else "N/A",
|
||
"Priority": total_priority,
|
||
"Hype Breakdown": {
|
||
"base": priority_comps["base"],
|
||
"time": priority_comps["time"],
|
||
"matchup_bonus": priority_comps["matchup_bonus"],
|
||
"score_state": priority_comps["score_state"],
|
||
"high_scoring": priority_comps["high_scoring"],
|
||
"power_play": priority_comps["power_play"],
|
||
"empty_net": priority_comps["empty_net"],
|
||
"comeback": momentum["comeback"],
|
||
"goal_spike": momentum["goal_spike"],
|
||
"importance": importance_comps["total"],
|
||
"importance_season_weight": importance_comps["season_weight"],
|
||
"importance_playoff_relevance": importance_comps[
|
||
"playoff_relevance"
|
||
],
|
||
"importance_rivalry": importance_comps["rivalry"],
|
||
"total": total_priority,
|
||
},
|
||
"Start Time": get_start_time(game),
|
||
"Start Time UTC": game.get("startTimeUTC", ""),
|
||
"Home Record": format_record(game["homeTeam"]["record"])
|
||
if game["gameState"] in ["PRE", "FUT"]
|
||
else "N/A",
|
||
"Away Record": format_record(game["awayTeam"]["record"])
|
||
if game["gameState"] in ["PRE", "FUT"]
|
||
else "N/A",
|
||
"Home Shots": game["homeTeam"]["sog"]
|
||
if game["gameState"] not in ["PRE", "FUT"]
|
||
else 0,
|
||
"Away Shots": game["awayTeam"]["sog"]
|
||
if game["gameState"] not in ["PRE", "FUT"]
|
||
else 0,
|
||
"Home Power Play": get_power_play_info(
|
||
game, game["homeTeam"]["name"]["default"]
|
||
),
|
||
"Away Power Play": get_power_play_info(
|
||
game, game["awayTeam"]["name"]["default"]
|
||
),
|
||
"Last Period Type": get_game_outcome(game, game_state),
|
||
}
|
||
)
|
||
|
||
def _sort_key(g):
|
||
if g["Game State"] == "PRE":
|
||
# Earliest start first — ISO-8601 sorts correctly as a string
|
||
return (0, g["Start Time UTC"], 0)
|
||
# LIVE / FINAL — highest priority first
|
||
return (1, "", -g["Priority"])
|
||
|
||
return sorted(extracted_info, key=_sort_key)
|
||
|
||
|
||
def _momentum_components(game):
|
||
"""Detects comeback recovery and fresh-goal spikes in a single pass.
|
||
|
||
Updates both caches exactly once per call. Returns:
|
||
- comeback: persistent bonus while a 2+ goal deficit is being recovered
|
||
- goal_spike: one-tick bonus on the refresh where a goal just landed
|
||
"""
|
||
zero = {"comeback": 0, "goal_spike": 0, "total": 0}
|
||
if game["gameState"] not in ("LIVE", "CRIT"):
|
||
return zero
|
||
if game["clock"]["inIntermission"]:
|
||
return zero
|
||
|
||
home_name = game["homeTeam"]["name"]["default"]
|
||
away_name = game["awayTeam"]["name"]["default"]
|
||
key = (home_name, away_name)
|
||
|
||
home_score = game["homeTeam"]["score"]
|
||
away_score = game["awayTeam"]["score"]
|
||
current = (home_score, away_score)
|
||
current_diff = abs(home_score - away_score)
|
||
period = game.get("periodDescriptor", {}).get("number", 0)
|
||
time_remaining = game.get("clock", {}).get("secondsRemaining", 0)
|
||
|
||
# ── Goal spike (fires the single tick after a score changes) ─────────
|
||
previous = _score_cache.get(key)
|
||
goal_spike = 0
|
||
if previous is not None and previous != current:
|
||
if period >= 4:
|
||
goal_spike = 100
|
||
elif period == 3 and time_remaining <= 300:
|
||
goal_spike = 80
|
||
elif period == 3 and time_remaining <= 720:
|
||
goal_spike = 60
|
||
elif period == 3:
|
||
goal_spike = 40
|
||
else:
|
||
goal_spike = 25
|
||
|
||
# ── Comeback tracking ────────────────────────────────────────────────
|
||
tracker_max = _comeback_tracker.get(key, 0)
|
||
if previous is not None:
|
||
tracker_max = max(tracker_max, abs(previous[0] - previous[1]))
|
||
_comeback_tracker[key] = tracker_max
|
||
_score_cache[key] = current
|
||
|
||
recovery = tracker_max - current_diff
|
||
comeback = 0
|
||
if recovery >= 2 and tracker_max >= 2:
|
||
base = {2: 50, 3: 90}.get(recovery, 120)
|
||
period_mult = {1: 0.6, 2: 0.8, 3: 1.0}.get(period, 1.2)
|
||
tie_bonus = 20 if current_diff == 0 else 0
|
||
comeback = int(base * period_mult + tie_bonus)
|
||
|
||
return {
|
||
"comeback": comeback,
|
||
"goal_spike": goal_spike,
|
||
"total": comeback + goal_spike,
|
||
}
|
||
|
||
|
||
def get_comeback_bonus(game):
|
||
return _momentum_components(game)["comeback"]
|
||
|
||
|
||
def get_goal_spike(game):
|
||
return _momentum_components(game)["goal_spike"]
|
||
|
||
|
||
def convert_game_state(game_state):
|
||
state_mapping = {"OFF": "FINAL", "CRIT": "LIVE", "FUT": "PRE"}
|
||
return state_mapping.get(game_state, game_state)
|
||
|
||
|
||
def get_period(game):
|
||
if game["gameState"] in ["PRE", "FUT"]:
|
||
return 0
|
||
elif game["gameState"] in ["FINAL", "OFF"]:
|
||
return "N/A"
|
||
else:
|
||
return game["periodDescriptor"]["number"]
|
||
|
||
|
||
def get_time_remaining(game):
|
||
if game["gameState"] in ["PRE", "FUT"]:
|
||
return "20:00"
|
||
elif game["gameState"] in ["FINAL", "OFF"]:
|
||
return "00:00"
|
||
else:
|
||
time_remaining = game["clock"]["timeRemaining"]
|
||
return "END" if time_remaining == "00:00" else time_remaining
|
||
|
||
|
||
def get_start_time(game):
|
||
if game["gameState"] in ["PRE", "FUT"]:
|
||
utc_time = game["startTimeUTC"]
|
||
est_time = utc_to_eastern(utc_time)
|
||
return est_time.lstrip("0")
|
||
else:
|
||
return "N/A"
|
||
|
||
|
||
def get_power_play_info(game, team_name):
|
||
situation = game.get("situation", {})
|
||
if not situation:
|
||
return ""
|
||
time_remaining = situation.get("timeRemaining", "")
|
||
home_descs = situation.get("homeTeam", {}).get("situationDescriptions", [])
|
||
away_descs = situation.get("awayTeam", {}).get("situationDescriptions", [])
|
||
if "PP" in home_descs and game["homeTeam"]["name"]["default"] == team_name:
|
||
return f"PP {time_remaining}"
|
||
if "PP" in away_descs and game["awayTeam"]["name"]["default"] == team_name:
|
||
return f"PP {time_remaining}"
|
||
return ""
|
||
|
||
|
||
def get_game_outcome(game, game_state):
|
||
return game["gameOutcome"]["lastPeriodType"] if game_state == "FINAL" else "N/A"
|
||
|
||
|
||
def _get_man_advantage(situation):
|
||
"""Parse situationCode for player count difference.
|
||
Format: [away_goalie][away_skaters][home_skaters][home_goalie]."""
|
||
code = situation.get("situationCode", "")
|
||
if len(code) != 4 or not code.isdigit():
|
||
return 1
|
||
away_total = int(code[0]) + int(code[1])
|
||
home_total = int(code[2]) + int(code[3])
|
||
return abs(home_total - away_total)
|
||
|
||
|
||
def _score_state_bonus(diff, period, time_remaining):
|
||
"""Unified score-state contribution: closeness reward AND blowout penalty,
|
||
coherent across period and time remaining. Replaces closeness + diff
|
||
penalty + late-3rd urgency as a single signal."""
|
||
if period >= 4:
|
||
# OT is always tied — flat tension bonus
|
||
return 60
|
||
|
||
if period <= 2:
|
||
return {0: 50, 1: 30, 2: 10, 3: -30, 4: -80}.get(diff, -120)
|
||
|
||
# Period 3 — depends on time remaining
|
||
mins_left = time_remaining / 60
|
||
if mins_left > 12:
|
||
return {0: 70, 1: 40, 2: 0, 3: -50, 4: -120}.get(diff, -160)
|
||
if mins_left > 6:
|
||
return {0: 120, 1: 80, 2: 20, 3: -80, 4: -180}.get(diff, -240)
|
||
if mins_left > 2:
|
||
# Goalie-pull zone — 2-goal deficit becomes interesting again
|
||
return {0: 180, 1: 140, 2: 50, 3: -80, 4: -200}.get(diff, -260)
|
||
# Final 2 min — peak tension
|
||
return {0: 220, 1: 180, 2: 80, 3: -50, 4: -220}.get(diff, -280)
|
||
|
||
|
||
def _priority_components(game):
|
||
"""Return a dict of all priority components plus the final total.
|
||
|
||
Calibrated so typical late-P3 tied games land around 550-650 and only
|
||
genuinely rare moments (playoff OT + comeback + PP) exceed 900.
|
||
"""
|
||
_zero = {
|
||
"base": 0,
|
||
"time": 0,
|
||
"matchup_bonus": 0,
|
||
"score_state": 0,
|
||
"high_scoring": 0,
|
||
"power_play": 0,
|
||
"empty_net": 0,
|
||
"total": 0,
|
||
}
|
||
|
||
if game["gameState"] in ["FINAL", "OFF", "PRE", "FUT"]:
|
||
return _zero
|
||
|
||
# Pushes intermission games to the bottom, retains relative sort order
|
||
time_remaining = game.get("clock", {}).get("secondsRemaining", 0)
|
||
if game["clock"]["inIntermission"]:
|
||
return {**_zero, "total": -2000 - time_remaining}
|
||
|
||
period = game.get("periodDescriptor", {}).get("number", 0)
|
||
home_score = game["homeTeam"]["score"]
|
||
away_score = game["awayTeam"]["score"]
|
||
score_difference = abs(home_score - away_score)
|
||
total_goals = home_score + away_score
|
||
is_playoff = game.get("gameType", 2) == 3
|
||
|
||
# ── 1. Base priority by period (tightened) ───────────────────────────
|
||
if is_playoff:
|
||
# Multi-OT escalates aggressively (P4=500, P5=620, P6=760, P7=920…)
|
||
base_priority = {1: 120, 2: 180, 3: 280}.get(
|
||
period, 500 + (period - 4) * 120 + max(0, period - 5) * 20
|
||
)
|
||
else:
|
||
# Regular season: P4=5-min OT, P5=shootout
|
||
base_priority = {1: 80, 2: 120, 3: 200, 4: 400, 5: 380}.get(period, 80)
|
||
|
||
# ── 2. Period length for time calculations ───────────────────────────
|
||
period_length = (1200 if is_playoff else 300) if period >= 4 else 1200
|
||
|
||
# ── 3. Matchup bonus (minor contributor now) ─────────────────────────
|
||
home_standings = get_team_standings(game["homeTeam"]["name"]["default"])
|
||
away_standings = get_team_standings(game["awayTeam"]["name"]["default"])
|
||
home_quality = (33 - home_standings["league_sequence"]) + (
|
||
33 - home_standings["league_l10_sequence"]
|
||
)
|
||
away_quality = (33 - away_standings["league_sequence"]) + (
|
||
33 - away_standings["league_l10_sequence"]
|
||
)
|
||
# Max combined quality = 128. Divided by 5 → 0-25.6. Multiplier tightens further.
|
||
matchup_raw = (home_quality + away_quality) / 5
|
||
matchup_multiplier = {1: 1.5, 2: 1.25, 3: 1.0, 4: 0.5}.get(period, 0.5)
|
||
matchup_bonus = int(matchup_raw * matchup_multiplier)
|
||
|
||
# ── Shootout: flat skills-competition score, no time component ──────
|
||
if period == 5 and not is_playoff:
|
||
so_total = int(380 + 60 + matchup_bonus)
|
||
return {
|
||
"base": 380,
|
||
"time": 0,
|
||
"matchup_bonus": matchup_bonus,
|
||
"score_state": 60,
|
||
"high_scoring": 0,
|
||
"power_play": 0,
|
||
"empty_net": 0,
|
||
"total": so_total,
|
||
}
|
||
|
||
# ── 4. Unified score-state bonus ─────────────────────────────────────
|
||
score_state = _score_state_bonus(score_difference, period, time_remaining)
|
||
|
||
# ── 5. Time priority (cap per period, non-linear toward end) ─────────
|
||
time_priority_max = {1: 30, 2: 60, 3: 120}.get(period, 200 if is_playoff else 60)
|
||
elapsed_fraction = (
|
||
max(0.0, (period_length - time_remaining) / period_length)
|
||
if period_length
|
||
else 0
|
||
)
|
||
time_priority = (elapsed_fraction**1.6) * time_priority_max
|
||
|
||
# ── 6. High-scoring bonus for close games with 6+ combined goals ─────
|
||
# Rewards the "shootout-y" vibe where next goal carries more weight.
|
||
high_scoring_bonus = 0
|
||
if total_goals >= 6 and score_difference <= 2:
|
||
high_scoring_bonus = min((total_goals - 5) * 12, 60)
|
||
if period < 3:
|
||
high_scoring_bonus = int(high_scoring_bonus * 0.5)
|
||
|
||
# ── 7. Power play bonus (tightened) ──────────────────────────────────
|
||
pp_bonus = 0
|
||
situation = game.get("situation", {})
|
||
home_descs = situation.get("homeTeam", {}).get("situationDescriptions", [])
|
||
away_descs = situation.get("awayTeam", {}).get("situationDescriptions", [])
|
||
if "PP" in home_descs or "PP" in away_descs:
|
||
man_advantage = _get_man_advantage(situation)
|
||
advantage_mult = 1.0 if man_advantage <= 1 else 1.5
|
||
if period >= 4:
|
||
pp_bonus = int(120 * advantage_mult)
|
||
elif period == 3 and time_remaining <= 300:
|
||
pp_bonus = int(90 * advantage_mult)
|
||
elif period == 3 and time_remaining <= 720:
|
||
pp_bonus = int(60 * advantage_mult)
|
||
elif period == 3:
|
||
pp_bonus = int(35 * advantage_mult)
|
||
else:
|
||
pp_bonus = int(20 * advantage_mult)
|
||
|
||
# ── 8. Empty net bonus (tightened) ───────────────────────────────────
|
||
en_bonus = 0
|
||
if "EN" in home_descs or "EN" in away_descs:
|
||
if period >= 4:
|
||
en_bonus = 180
|
||
elif period == 3 and time_remaining <= 180:
|
||
en_bonus = 140
|
||
elif period == 3 and time_remaining <= 360:
|
||
en_bonus = 100
|
||
else:
|
||
en_bonus = 50
|
||
|
||
logger.debug(
|
||
"priority components — base: %s, time: %.0f, matchup: %s, "
|
||
"score_state: %s, high_scoring: %s, pp: %s, en: %s",
|
||
base_priority,
|
||
time_priority,
|
||
matchup_bonus,
|
||
score_state,
|
||
high_scoring_bonus,
|
||
pp_bonus,
|
||
en_bonus,
|
||
)
|
||
|
||
final_priority = int(
|
||
base_priority
|
||
+ time_priority
|
||
+ matchup_bonus
|
||
+ score_state
|
||
+ high_scoring_bonus
|
||
+ pp_bonus
|
||
+ en_bonus
|
||
)
|
||
|
||
return {
|
||
"base": base_priority,
|
||
"time": int(time_priority),
|
||
"matchup_bonus": matchup_bonus,
|
||
"score_state": score_state,
|
||
"high_scoring": high_scoring_bonus,
|
||
"power_play": pp_bonus,
|
||
"empty_net": en_bonus,
|
||
"total": final_priority,
|
||
}
|
||
|
||
|
||
def calculate_game_priority(game):
|
||
return _priority_components(game)["total"]
|
||
|
||
|
||
def get_team_standings(team_name):
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"""
|
||
SELECT league_sequence, league_l10_sequence,
|
||
division_abbrev, conference_abbrev,
|
||
games_played, wildcard_sequence
|
||
FROM standings
|
||
WHERE team_common_name = ?
|
||
""",
|
||
(team_name,),
|
||
)
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
if result:
|
||
return {
|
||
"league_sequence": result[0],
|
||
"league_l10_sequence": result[1],
|
||
"division_abbrev": result[2],
|
||
"conference_abbrev": result[3],
|
||
"games_played": result[4],
|
||
"wildcard_sequence": result[5],
|
||
}
|
||
return {
|
||
"league_sequence": 0,
|
||
"league_l10_sequence": 0,
|
||
"division_abbrev": None,
|
||
"conference_abbrev": None,
|
||
"games_played": 0,
|
||
"wildcard_sequence": 32,
|
||
}
|
||
|
||
|
||
def _playoff_importance(game):
|
||
"""Importance for playoff games — round + series context. Max 150."""
|
||
series = game.get("seriesStatus", {})
|
||
if not series:
|
||
return {
|
||
"season_weight": 1.0,
|
||
"playoff_relevance": 0.50,
|
||
"rivalry": 1.0,
|
||
"total": 60,
|
||
}
|
||
|
||
round_num = series.get("round", 1)
|
||
top_wins = series.get("topSeedWins", 0)
|
||
bottom_wins = series.get("bottomSeedWins", 0)
|
||
max_wins = max(top_wins, bottom_wins)
|
||
min_wins = min(top_wins, bottom_wins)
|
||
|
||
round_mult = {1: 1.0, 2: 1.15, 3: 1.30, 4: 1.50}.get(round_num, 1.0)
|
||
|
||
if max_wins == 3 and min_wins == 3:
|
||
series_factor = 1.0 # Game 7
|
||
elif max_wins == 3:
|
||
series_factor = 0.90 # Elimination game
|
||
elif max_wins == 2 and min_wins == 2:
|
||
series_factor = 0.75 # Pivotal tied series
|
||
elif max_wins == 2:
|
||
series_factor = 0.60
|
||
else:
|
||
series_factor = 0.45
|
||
|
||
importance = min(int(series_factor * round_mult * 100), 150)
|
||
|
||
return {
|
||
"season_weight": round_mult,
|
||
"playoff_relevance": series_factor,
|
||
"rivalry": 1.0,
|
||
"total": importance,
|
||
}
|
||
|
||
|
||
def _importance_components(game):
|
||
"""Regular-season importance — season_weight × stakes × rivalry. Max 100."""
|
||
_zero = {"season_weight": 0.0, "playoff_relevance": 0.0, "rivalry": 1.0, "total": 0}
|
||
|
||
if game["gameState"] in ("FINAL", "OFF"):
|
||
return _zero
|
||
if game.get("gameType", 2) == 3:
|
||
return _playoff_importance(game)
|
||
if game.get("gameType", 2) != 2:
|
||
return _zero
|
||
|
||
home_st = get_team_standings(game["homeTeam"]["name"]["default"])
|
||
away_st = get_team_standings(game["awayTeam"]["name"]["default"])
|
||
|
||
# Season weight — zero before game 30, linear ramp to 1.0 at game 70+
|
||
avg_gp = (home_st["games_played"] + away_st["games_played"]) / 2
|
||
if avg_gp <= 30:
|
||
season_weight = 0.0
|
||
else:
|
||
season_weight = min((avg_gp - 30) / 40, 1.0)
|
||
|
||
# Playoff stakes — peaks on the bubble, drops for locked-in or out
|
||
best_wc = min(
|
||
home_st["wildcard_sequence"] or 32, away_st["wildcard_sequence"] or 32
|
||
)
|
||
if best_wc <= 8:
|
||
stakes = 0.5 # locked in, pressure off
|
||
elif best_wc <= 16:
|
||
stakes = 0.8 # comfortable, meaningful
|
||
elif best_wc <= 20:
|
||
stakes = 1.0 # bubble, every point critical
|
||
elif best_wc <= 24:
|
||
stakes = 0.5 # slipping
|
||
else:
|
||
stakes = 0.2 # effectively out
|
||
|
||
# Rivalry — division > conference > other
|
||
home_div = home_st["division_abbrev"]
|
||
away_div = away_st["division_abbrev"]
|
||
home_conf = home_st["conference_abbrev"]
|
||
away_conf = away_st["conference_abbrev"]
|
||
if home_div and away_div and home_div == away_div:
|
||
rivalry_multiplier = 1.4
|
||
elif home_conf and away_conf and home_conf == away_conf:
|
||
rivalry_multiplier = 1.2
|
||
else:
|
||
rivalry_multiplier = 1.0
|
||
|
||
importance = int(season_weight * stakes * rivalry_multiplier * 70)
|
||
importance = max(0, min(importance, 100))
|
||
|
||
logger.debug(
|
||
"importance — season_weight: %.2f, stakes: %.2f, rivalry: %.1f, total: %s",
|
||
season_weight,
|
||
stakes,
|
||
rivalry_multiplier,
|
||
importance,
|
||
)
|
||
|
||
return {
|
||
"season_weight": round(season_weight, 3),
|
||
"playoff_relevance": stakes,
|
||
"rivalry": rivalry_multiplier,
|
||
"total": importance,
|
||
}
|
||
|
||
|
||
def calculate_game_importance(game):
|
||
return _importance_components(game)["total"]
|
||
|
||
|
||
def utc_to_eastern(utc_time):
|
||
utc_datetime = datetime.strptime(utc_time, "%Y-%m-%dT%H:%M:%SZ")
|
||
eastern_datetime = utc_datetime.replace(tzinfo=timezone.utc).astimezone(EASTERN)
|
||
return eastern_datetime.strftime("%I:%M %p")
|