cal/major-domo-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Draft pick service and draft helpers

Browse Source
This commit is contained in:
Cal Corum 2025-10-24 10:24:14 -05:00
parent 0e676c86fd
commit 86459693a4
3 changed files with 545 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -219,3 +219,4 @@ __marimo__/
# Project-specific
data/
production_logs/
*.json

312
services/draft_pick_service.py Normal file
View File

@ -0,0 +1,312 @@
"""
Draft pick service for Discord Bot v2.0
Handles draft pick CRUD operations. NO CACHING - draft data changes constantly.
"""
import logging
from typing import Optional, List
from services.base_service import BaseService
from models.draft_pick import DraftPick
from exceptions import APIException
logger = logging.getLogger(f'{__name__}.DraftPickService')
class DraftPickService(BaseService[DraftPick]):
"""
Service for draft pick operations.
IMPORTANT: This service does NOT use caching decorators because draft picks
change constantly during an active draft. Always fetch fresh data.
Features:
- Get pick by overall number
- Get picks by team
- Get picks by round
- Update pick with player selection
- Query available/taken picks
"""
def __init__(self):
"""Initialize draft pick service."""
super().__init__(DraftPick, 'draftpicks')
logger.debug("DraftPickService initialized")
async def get_pick(self, season: int, overall: int) -> Optional[DraftPick]:
"""
Get specific pick by season and overall number.
NOT cached - picks change during draft.
Args:
season: Draft season
overall: Overall pick number
Returns:
DraftPick instance or None if not found
"""
try:
params = [
('season', str(season)),
('overall', str(overall))
]
picks = await self.get_all_items(params=params)
if picks:
pick = picks[0]
logger.debug(f"Found pick #{overall} for season {season}")
return pick
logger.debug(f"No pick found for season {season}, overall #{overall}")
return None
except Exception as e:
logger.error(f"Error getting pick season={season} overall={overall}: {e}")
return None
async def get_picks_by_team(
self,
season: int,
team_id: int,
round_start: int = 1,
round_end: int = 32
) -> List[DraftPick]:
"""
Get all picks owned by a team in a season.
NOT cached - picks change as they're traded.
Args:
season: Draft season
team_id: Team ID that owns the picks
round_start: Starting round (inclusive)
round_end: Ending round (inclusive)
Returns:
List of DraftPick instances owned by team
"""
try:
params = [
('season', str(season)),
('owner_team_id', str(team_id)),
('round_start', str(round_start)),
('round_end', str(round_end)),
('sort', 'order-asc')
]
picks = await self.get_all_items(params=params)
logger.debug(f"Found {len(picks)} picks for team {team_id} in rounds {round_start}-{round_end}")
return picks
except Exception as e:
logger.error(f"Error getting picks for team {team_id}: {e}")
return []
async def get_picks_by_round(
self,
season: int,
round_num: int,
include_taken: bool = True
) -> List[DraftPick]:
"""
Get all picks in a specific round.
NOT cached - picks change as they're selected.
Args:
season: Draft season
round_num: Round number
include_taken: Whether to include picks with players selected
Returns:
List of DraftPick instances in the round
"""
try:
params = [
('season', str(season)),
('pick_round_start', str(round_num)),
('pick_round_end', str(round_num)),
('sort', 'order-asc')
]
if not include_taken:
params.append(('player_taken', 'false'))
picks = await self.get_all_items(params=params)
logger.debug(f"Found {len(picks)} picks in round {round_num}")
return picks
except Exception as e:
logger.error(f"Error getting picks for round {round_num}: {e}")
return []
async def get_available_picks(
self,
season: int,
overall_start: Optional[int] = None,
overall_end: Optional[int] = None
) -> List[DraftPick]:
"""
Get picks that haven't been selected yet.
NOT cached - availability changes constantly.
Args:
season: Draft season
overall_start: Starting overall pick number (optional)
overall_end: Ending overall pick number (optional)
Returns:
List of available DraftPick instances
"""
try:
params = [
('season', str(season)),
('player_taken', 'false'),
('sort', 'order-asc')
]
if overall_start is not None:
params.append(('overall_start', str(overall_start)))
if overall_end is not None:
params.append(('overall_end', str(overall_end)))
picks = await self.get_all_items(params=params)
logger.debug(f"Found {len(picks)} available picks")
return picks
except Exception as e:
logger.error(f"Error getting available picks: {e}")
return []
async def get_recent_picks(
self,
season: int,
overall_end: int,
limit: int = 5
) -> List[DraftPick]:
"""
Get recent picks before a specific pick number.
NOT cached - recent picks change as draft progresses.
Args:
season: Draft season
overall_end: Get picks before this overall number
limit: Number of picks to retrieve
Returns:
List of recent DraftPick instances (reverse chronological)
"""
try:
params = [
('season', str(season)),
('overall_end', str(overall_end - 1)), # Exclude current pick
('player_taken', 'true'), # Only taken picks
('sort', 'order-desc'), # Most recent first
('limit', str(limit))
]
picks = await self.get_all_items(params=params)
logger.debug(f"Found {len(picks)} recent picks before #{overall_end}")
return picks
except Exception as e:
logger.error(f"Error getting recent picks: {e}")
return []
async def get_upcoming_picks(
self,
season: int,
overall_start: int,
limit: int = 5
) -> List[DraftPick]:
"""
Get upcoming picks after a specific pick number.
NOT cached - upcoming picks change as draft progresses.
Args:
season: Draft season
overall_start: Get picks after this overall number
limit: Number of picks to retrieve
Returns:
List of upcoming DraftPick instances
"""
try:
params = [
('season', str(season)),
('overall_start', str(overall_start + 1)), # Exclude current pick
('sort', 'order-asc'), # Chronological order
('limit', str(limit))
]
picks = await self.get_all_items(params=params)
logger.debug(f"Found {len(picks)} upcoming picks after #{overall_start}")
return picks
except Exception as e:
logger.error(f"Error getting upcoming picks: {e}")
return []
async def update_pick_selection(
self,
pick_id: int,
player_id: int
) -> Optional[DraftPick]:
"""
Update a pick with player selection.
Args:
pick_id: Draft pick database ID
player_id: Player ID being selected
Returns:
Updated DraftPick instance or None if update failed
"""
try:
update_data = {'player_id': player_id}
updated_pick = await self.patch(pick_id, update_data)
if updated_pick:
logger.info(f"Updated pick #{pick_id} with player {player_id}")
else:
logger.error(f"Failed to update pick #{pick_id}")
return updated_pick
except Exception as e:
logger.error(f"Error updating pick {pick_id}: {e}")
return None
async def clear_pick_selection(self, pick_id: int) -> Optional[DraftPick]:
"""
Clear player selection from a pick (for admin wipe operations).
Args:
pick_id: Draft pick database ID
Returns:
Updated DraftPick instance with player cleared, or None if failed
"""
try:
update_data = {'player_id': None}
updated_pick = await self.patch(pick_id, update_data)
if updated_pick:
logger.info(f"Cleared player selection from pick #{pick_id}")
else:
logger.error(f"Failed to clear pick #{pick_id}")
return updated_pick
except Exception as e:
logger.error(f"Error clearing pick {pick_id}: {e}")
return None
# Global service instance
draft_pick_service = DraftPickService()

232
utils/draft_helpers.py Normal file
View File

@ -0,0 +1,232 @@
"""
Draft utility functions for Discord Bot v2.0
Provides helper functions for draft order calculation and cap space validation.
"""
import math
from typing import Tuple
from utils.logging import get_contextual_logger
logger = get_contextual_logger(__name__)
def calculate_pick_details(overall: int) -> Tuple[int, int]:
"""
Calculate round number and pick position from overall pick number.
Hybrid draft format:
- Rounds 1-10: Linear (same order every round)
- Rounds 11+: Snake (reverse order on even rounds)
Special rule: Round 11, Pick 1 belongs to the team that had Round 10, Pick 16
(last pick of linear rounds transitions to first pick of snake rounds).
Args:
overall: Overall pick number (1-512 for 32-round, 16-team draft)
Returns:
(round_num, position): Round number (1-32) and position within round (1-16)
Examples:
>>> calculate_pick_details(1)
(1, 1) # Round 1, Pick 1
>>> calculate_pick_details(16)
(1, 16) # Round 1, Pick 16
>>> calculate_pick_details(160)
(10, 16) # Round 10, Pick 16 (last linear pick)
>>> calculate_pick_details(161)
(11, 1) # Round 11, Pick 1 (first snake pick - same team as 160)
>>> calculate_pick_details(176)
(11, 16) # Round 11, Pick 16
>>> calculate_pick_details(177)
(12, 16) # Round 12, Pick 16 (snake reverses)
"""
round_num = math.ceil(overall / 16)
if round_num <= 10:
# Linear draft: position is same calculation every round
position = ((overall - 1) % 16) + 1
else:
# Snake draft: reverse on even rounds
if round_num % 2 == 1: # Odd rounds (11, 13, 15...)
position = ((overall - 1) % 16) + 1
else: # Even rounds (12, 14, 16...)
position = 16 - ((overall - 1) % 16)
return round_num, position
def calculate_overall_from_round_position(round_num: int, position: int) -> int:
"""
Calculate overall pick number from round and position.
Inverse operation of calculate_pick_details().
Args:
round_num: Round number (1-32)
position: Position within round (1-16)
Returns:
Overall pick number
Examples:
>>> calculate_overall_from_round_position(1, 1)
1
>>> calculate_overall_from_round_position(10, 16)
160
>>> calculate_overall_from_round_position(11, 1)
161
>>> calculate_overall_from_round_position(12, 16)
177
"""
if round_num <= 10:
# Linear draft
return (round_num - 1) * 16 + position
else:
# Snake draft
picks_before_round = (round_num - 1) * 16
if round_num % 2 == 1: # Odd snake rounds
return picks_before_round + position
else: # Even snake rounds (reversed)
return picks_before_round + (17 - position)
async def validate_cap_space(
roster: dict,
new_player_wara: float
) -> Tuple[bool, float]:
"""
Validate team has cap space to draft player.
Cap calculation:
- Maximum 32 players on active roster
- Only top 26 players count toward cap
- Cap limit: 32.00 sWAR total
Args:
roster: Roster dictionary from API with structure:
{
'active': {
'players': [{'id': int, 'name': str, 'wara': float}, ...],
'WARa': float # Current roster sWAR
}
}
new_player_wara: sWAR value of player being drafted
Returns:
(valid, projected_total): True if under cap, projected total sWAR after addition
Raises:
ValueError: If roster structure is invalid
"""
if not roster or not roster.get('active'):
raise ValueError("Invalid roster structure - missing 'active' key")
active_roster = roster['active']
current_players = active_roster.get('players', [])
# Calculate how many players count toward cap after adding new player
current_roster_size = len(current_players)
projected_roster_size = current_roster_size + 1
# Maximum zeroes = 32 - roster size
# Maximum counted = 26 - zeroes
max_zeroes = 32 - projected_roster_size
max_counted = min(26, 26 - max_zeroes) # Can't count more than 26
# Sort all players (including new) by sWAR descending
all_players_wara = [p['wara'] for p in current_players] + [new_player_wara]
sorted_wara = sorted(all_players_wara, reverse=True)
# Sum top N players
projected_total = sum(sorted_wara[:max_counted])
# Allow tiny floating point tolerance
is_valid = projected_total <= 32.00001
logger.debug(
f"Cap validation: roster_size={current_roster_size}, "
f"projected_size={projected_roster_size}, "
f"max_counted={max_counted}, "
f"new_player_wara={new_player_wara:.2f}, "
f"projected_total={projected_total:.2f}, "
f"valid={is_valid}"
)
return is_valid, projected_total
def format_pick_display(overall: int) -> str:
"""
Format pick number for display.
Args:
overall: Overall pick number
Returns:
Formatted string like "Round 1, Pick 3 (Overall #3)"
Examples:
>>> format_pick_display(1)
"Round 1, Pick 1 (Overall #1)"
>>> format_pick_display(45)
"Round 3, Pick 13 (Overall #45)"
"""
round_num, position = calculate_pick_details(overall)
return f"Round {round_num}, Pick {position} (Overall #{overall})"
def get_next_pick_overall(current_overall: int) -> int:
"""
Get the next overall pick number.
Simply increments by 1, but provided for completeness and future logic changes.
Args:
current_overall: Current overall pick number
Returns:
Next overall pick number
"""
return current_overall + 1
def is_draft_complete(current_overall: int, total_picks: int = 512) -> bool:
"""
Check if draft is complete.
Args:
current_overall: Current overall pick number
total_picks: Total number of picks in draft (default: 512 for 32 rounds, 16 teams)
Returns:
True if draft is complete
"""
return current_overall > total_picks
def get_round_name(round_num: int) -> str:
"""
Get display name for round.
Args:
round_num: Round number
Returns:
Display name like "Round 1" or "Round 11 (Snake Draft Begins)"
"""
if round_num == 1:
return "Round 1"
elif round_num == 11:
return "Round 11 (Snake Draft Begins)"
else:
return f"Round {round_num}"