josh
7784eaf9ce
- Empty net detection: pulled goalie triggers +150-250 bonus, stacks with PP - Playoff series importance: Game 7 Cup Final = 200, elimination games scale by round; fallback of 100 when series data unavailable - Period-aware score differential: 2-goal deficit penalty DECREASES in final 2 min of P3 (goalie-pull zone), 3+ goal games get harsher penalties late - Persistent comeback narrative: tracks max deficit, sustained bonus for 2+ goal recoveries instead of one-shot spike (0-3 to 3-3 = 150 persistent) - Shootout special handling: flat base 550 with no time component; ranks below dramatic close P3 games (skills competition, not hockey) - Multi-man advantage: parses situationCode for 5v3/4v3, applies 1.6x PP mult - Non-linear time priority: elapsed^1.5 curve weights final minutes more - Matchup multiplier rebalance: P1/P2 from 2.0/1.65 to 1.5, tiebreaker not dominant factor - Frontend gauge max raised from 700 to 1000 with adjusted color thresholds Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
541 lines
19 KiB
Python
541 lines
19 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)
|
|
comeback = get_comeback_bonus(game)
|
|
importance_comps = _importance_components(game)
|
|
total_priority = priority_comps["total"] + comeback + 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"],
|
|
"closeness": priority_comps["closeness"],
|
|
"power_play": priority_comps["power_play"],
|
|
"empty_net": priority_comps["empty_net"],
|
|
"comeback": comeback,
|
|
"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),
|
|
"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),
|
|
}
|
|
)
|
|
|
|
# Sort games based on priority
|
|
return sorted(extracted_info, key=lambda x: x["Priority"], reverse=True)
|
|
|
|
|
|
def get_comeback_bonus(game):
|
|
"""Persistent comeback bonus that scales with deficit recovered.
|
|
|
|
Tracks the maximum score differential seen in the game. A recovery of 2+
|
|
goals earns a sustained bonus that persists as long as the game remains
|
|
close. One-goal swings are normal hockey and earn no bonus.
|
|
"""
|
|
if game["gameState"] not in ("LIVE", "CRIT"):
|
|
return 0
|
|
if game["clock"]["inIntermission"]:
|
|
return 0
|
|
|
|
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_diff = abs(home_score - away_score)
|
|
period = game.get("periodDescriptor", {}).get("number", 0)
|
|
|
|
tracker_max = _comeback_tracker.get(key, 0)
|
|
if key in _score_cache:
|
|
prev_diff = abs(_score_cache[key][0] - _score_cache[key][1])
|
|
tracker_max = max(tracker_max, prev_diff)
|
|
_comeback_tracker[key] = tracker_max
|
|
_score_cache[key] = (home_score, away_score)
|
|
|
|
recovery = tracker_max - current_diff
|
|
if recovery < 2 or tracker_max < 2:
|
|
return 0
|
|
|
|
base = {2: 60, 3: 120}.get(recovery, 160)
|
|
period_mult = {1: 0.6, 2: 0.8, 3: 1.0}.get(period, 1.2)
|
|
tie_bonus = 30 if current_diff == 0 else 0
|
|
|
|
return int(base * period_mult + tie_bonus)
|
|
|
|
|
|
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 _priority_components(game):
|
|
"""Return a dict of all priority components plus the final total."""
|
|
_zero = {
|
|
"base": 0,
|
|
"time": 0,
|
|
"matchup_bonus": 0,
|
|
"closeness": 0,
|
|
"power_play": 0,
|
|
"empty_net": 0,
|
|
"total": 0,
|
|
}
|
|
|
|
if game["gameState"] in ["FINAL", "OFF", "PRE", "FUT"]:
|
|
return _zero
|
|
|
|
period = game.get("periodDescriptor", {}).get("number", 0)
|
|
time_remaining = game.get("clock", {}).get("secondsRemaining", 0)
|
|
home_score = game["homeTeam"]["score"]
|
|
away_score = game["awayTeam"]["score"]
|
|
score_difference = abs(home_score - away_score)
|
|
is_playoff = game.get("gameType", 2) == 3
|
|
|
|
# ── 1. Base priority by period ────────────────────────────────────────
|
|
if is_playoff:
|
|
base_priority = {1: 150, 2: 200, 3: 350}.get(period, 600 + (period - 4) * 150)
|
|
else:
|
|
base_priority = {1: 150, 2: 200, 3: 350, 4: 600, 5: 700}.get(period, 150)
|
|
|
|
# ── 2. Period length for time calculations ────────────────────────────
|
|
period_length = (1200 if is_playoff else 300) if period >= 4 else 1200
|
|
|
|
# ── 3. Standings-quality matchup bonus ───────────────────────────────
|
|
# Invert rank so that #1 (best) contributes the most quality points.
|
|
# league_sequence 1=best, 32=worst → inverted: 32 quality pts for #1, 1 for #32.
|
|
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"]
|
|
)
|
|
# Higher period = matchup matters less (any OT is exciting regardless of teams)
|
|
matchup_multiplier = {1: 1.5, 2: 1.5, 3: 1.25, 4: 1.0}.get(period, 1.0)
|
|
matchup_bonus = (home_quality + away_quality) * matchup_multiplier
|
|
|
|
# ── Shootout: flat priority, no time component (rounds, not clock) ───
|
|
if period == 5 and not is_playoff:
|
|
so_base = 550
|
|
so_closeness = 80
|
|
so_matchup = (home_quality + away_quality) * 1.0
|
|
so_total = int(so_base + so_closeness + so_matchup)
|
|
return {
|
|
"base": so_base,
|
|
"time": 0,
|
|
"matchup_bonus": int(so_matchup),
|
|
"closeness": so_closeness,
|
|
"power_play": 0,
|
|
"empty_net": 0,
|
|
"total": so_total,
|
|
}
|
|
|
|
# ── 4. Score-differential penalty (period-aware) ───────────────────────
|
|
score_differential_adjustment = 0
|
|
if period <= 2:
|
|
adj = {0: 0, 1: 0, 2: 60, 3: 200, 4: 350}
|
|
score_differential_adjustment = adj.get(
|
|
score_difference, 350 + (score_difference - 4) * 100
|
|
)
|
|
elif period == 3:
|
|
mins_left = time_remaining / 60
|
|
if mins_left > 10:
|
|
adj = {0: 0, 1: 0, 2: 80, 3: 250, 4: 400}
|
|
elif mins_left > 5:
|
|
adj = {0: 0, 1: 0, 2: 120, 3: 350, 4: 500}
|
|
elif mins_left > 2:
|
|
# Goalie-pull zone: 2-goal penalty DECREASES
|
|
adj = {0: 0, 1: 0, 2: 80, 3: 450, 4: 600}
|
|
else:
|
|
# Final 2 min: 2-goal deficit with active goalie pull is exciting
|
|
adj = {0: 0, 1: 0, 2: 60, 3: 550, 4: 700}
|
|
score_differential_adjustment = adj.get(
|
|
score_difference, adj[4] + (score_difference - 4) * 100
|
|
)
|
|
# OT: always tied, no penalty needed
|
|
|
|
base_priority -= score_differential_adjustment
|
|
|
|
# ── 5. Late-3rd urgency bonus ─────────────────────────────────────────
|
|
if period == 3 and time_remaining <= 720:
|
|
if score_difference == 0:
|
|
base_priority += 100
|
|
elif score_difference == 1:
|
|
base_priority += 60
|
|
|
|
if period == 3 and time_remaining <= 360:
|
|
if score_difference == 0:
|
|
base_priority += 50
|
|
elif score_difference == 1:
|
|
base_priority += 30
|
|
|
|
# ── 6. Closeness bonus ───────────────────────────────────────────────
|
|
closeness_bonus = {0: 80, 1: 50, 2: 20}.get(score_difference, 0)
|
|
|
|
# ── 7. Time priority (non-linear — final minutes weighted more) ─────
|
|
time_multiplier = {4: 5, 3: 5, 2: 3}.get(period, 2.0 if period >= 5 else 1.5)
|
|
elapsed_fraction = (
|
|
max(0.0, (period_length - time_remaining) / period_length)
|
|
if period_length
|
|
else 0
|
|
)
|
|
time_priority = (elapsed_fraction**1.5) * (period_length / 20) * time_multiplier
|
|
|
|
# ── 8. Power play bonus ───────────────────────────────────────────────
|
|
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.6
|
|
if period >= 4:
|
|
pp_bonus = int(200 * advantage_mult)
|
|
elif period == 3 and time_remaining <= 300:
|
|
pp_bonus = int(150 * advantage_mult)
|
|
elif period == 3 and time_remaining <= 720:
|
|
pp_bonus = int(100 * advantage_mult)
|
|
elif period == 3:
|
|
pp_bonus = int(50 * advantage_mult)
|
|
else:
|
|
pp_bonus = int(30 * advantage_mult)
|
|
|
|
# ── 9. Empty net bonus ───────────────────────────────────────────────
|
|
en_bonus = 0
|
|
if "EN" in home_descs or "EN" in away_descs:
|
|
if period >= 4:
|
|
en_bonus = 250
|
|
elif period == 3 and time_remaining <= 180:
|
|
en_bonus = 200
|
|
elif period == 3 and time_remaining <= 360:
|
|
en_bonus = 150
|
|
else:
|
|
en_bonus = 75
|
|
|
|
logger.debug(
|
|
"priority components — base: %s, time: %.0f, matchup_bonus: %.0f, "
|
|
"closeness: %s, pp: %s, en: %s",
|
|
base_priority,
|
|
time_priority,
|
|
matchup_bonus,
|
|
closeness_bonus,
|
|
pp_bonus,
|
|
en_bonus,
|
|
)
|
|
|
|
final_priority = int(
|
|
base_priority
|
|
+ time_priority
|
|
+ matchup_bonus
|
|
+ closeness_bonus
|
|
+ pp_bonus
|
|
+ en_bonus
|
|
)
|
|
|
|
# Pushes intermission games to the bottom, retains relative sort order
|
|
if game["clock"]["inIntermission"]:
|
|
return {**_zero, "total": -2000 - time_remaining}
|
|
|
|
return {
|
|
"base": base_priority,
|
|
"time": int(time_priority),
|
|
"matchup_bonus": int(matchup_bonus),
|
|
"closeness": closeness_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 based on series context and round."""
|
|
series = game.get("seriesStatus", {})
|
|
if not series:
|
|
# No series data available — flat playoff bonus
|
|
return {
|
|
"season_weight": 1.0,
|
|
"playoff_relevance": 0.50,
|
|
"rivalry": 1.0,
|
|
"total": 100,
|
|
}
|
|
|
|
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
|
|
elif max_wins == 3:
|
|
series_factor = 0.85
|
|
elif max_wins == 2 and min_wins == 2:
|
|
series_factor = 0.70
|
|
elif max_wins == 2:
|
|
series_factor = 0.55
|
|
else:
|
|
series_factor = 0.40
|
|
|
|
importance = min(int(series_factor * round_mult * 200), 200)
|
|
|
|
return {
|
|
"season_weight": round_mult,
|
|
"playoff_relevance": series_factor,
|
|
"rivalry": 1.0,
|
|
"total": importance,
|
|
}
|
|
|
|
|
|
def _importance_components(game):
|
|
"""Return a dict of all importance components plus the final total."""
|
|
_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 — near-zero before game 30, sharp ramp 55-70, max at 82
|
|
avg_gp = (home_st["games_played"] + away_st["games_played"]) / 2
|
|
if avg_gp <= 30:
|
|
season_weight = 0.05
|
|
else:
|
|
t = (avg_gp - 30) / (82 - 30)
|
|
season_weight = min(t**1.8, 1.0)
|
|
|
|
# Playoff relevance — peaks for bubble teams (wildcard rank ~17-19)
|
|
best_wc = min(
|
|
home_st["wildcard_sequence"] or 32, away_st["wildcard_sequence"] or 32
|
|
)
|
|
if best_wc <= 12:
|
|
playoff_relevance = 0.60
|
|
elif best_wc <= 16:
|
|
playoff_relevance = 0.85
|
|
elif best_wc <= 19:
|
|
playoff_relevance = 1.00
|
|
elif best_wc <= 23:
|
|
playoff_relevance = 0.65
|
|
else:
|
|
playoff_relevance = 0.15
|
|
|
|
# Division/conference rivalry multiplier
|
|
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
|
|
|
|
raw = season_weight * playoff_relevance * rivalry_multiplier
|
|
importance = max(0, min(int((raw / 1.4) * 150), 150))
|
|
|
|
logger.debug(
|
|
"importance components — season_weight: %.3f, playoff_relevance: %.2f, "
|
|
"rivalry: %.1f, importance: %s",
|
|
season_weight,
|
|
playoff_relevance,
|
|
rivalry_multiplier,
|
|
importance,
|
|
)
|
|
|
|
return {
|
|
"season_weight": round(season_weight, 3),
|
|
"playoff_relevance": playoff_relevance,
|
|
"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")
|