cal/sba-scouting
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Lineup Builder, Gameday screen, and matchup scoring system

Browse Source 
Features:
- Lineup Builder screen: set batting order, assign positions, save/load lineups
- Gameday screen: integrated matchup scout + lineup builder side-by-side
- Matchup Scout: analyze batters vs opposing pitchers with standardized scoring
- Standardized scoring system with league AVG/STDEV calculations
- Score caching for fast matchup lookups

Lineup Builder (press 'l'):
- Dual-panel UI with available batters and 9-slot lineup
- Keyboard controls: a=add, r=remove, k/j=reorder, p=change position
- Save/load named lineups, delete saved lineups with 'd'

Gameday screen (press 'g'):
- Left panel: team/pitcher selection with matchup ratings
- Right panel: lineup builder with live matchup ratings per batter
- Players in lineup marked with * in matchup list
- Click highlighted row to toggle selection for screenshots

Other changes:
- Dynamic season configuration (removed hardcoded season=13)
- Added delete_lineup query function
- StandardizedScoreCache model for pre-computed scores
- Auto-rebuild score cache after card imports
This commit is contained in:
Cal Corum 2026-01-25 14:09:22 -06:00
parent 5b307a91a6
commit 3c76ce1cf0
14 changed files with 3620 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

448
PROJECT_PLAN.json Normal file
View File

@ -0,0 +1,448 @@
{
"meta": {
"version": "1.0.0",
"created": "2026-01-25",
"lastUpdated": "2026-01-25",
"planType": "feature",
"project": "SBA Scout TUI",
"description": "Fantasy baseball scouting TUI application for the SBA (Strat-o-Matic Baseball Association) league",
"totalEstimatedHours": 40,
"totalTasks": 18,
"completedTasks": 7
},
"categories": {
"critical": "Must fix immediately - blocks core functionality",
"high": "Required for production-ready app",
"medium": "Quality improvements and secondary features",
"low": "Polish, nice-to-have features",
"feature": "New capabilities and screens"
},
"tasks": [
{
"id": "FEAT-001",
"name": "Dashboard Screen",
"description": "Main dashboard showing roster summary (Majors/Minors/IL counts), sWAR usage, and quick navigation buttons",
"category": "feature",
"priority": 1,
"completed": true,
"tested": true,
"dependencies": [],
"files": [
{
"path": "src/sba_scout/app.py",
"lines": [30, 180],
"issue": "DashboardScreen class implementation"
}
],
"suggestedFix": null,
"estimatedHours": 4,
"notes": "Completed - shows roster counts and sWAR from settings"
},
{
"id": "FEAT-002",
"name": "Roster Screen",
"description": "Full roster view with tabbed display (Majors/Minors/IL), separate DataTables for batters and pitchers with ratings and stats",
"category": "feature",
"priority": 2,
"completed": true,
"tested": true,
"dependencies": ["FEAT-001"],
"files": [
{
"path": "src/sba_scout/screens/roster.py",
"lines": [1, 270],
"issue": "RosterScreen implementation"
}
],
"suggestedFix": null,
"estimatedHours": 6,
"notes": "Completed - batters show vL/vR/Ovr/Defense, pitchers show vL/vR/Ovr/S/R/C endurance"
},
{
"id": "FEAT-003",
"name": "Card Data Importer",
"description": "Import batter and pitcher card data from BatterCalcs.csv and PitcherCalcs.csv exports, including pre-calculated ratings",
"category": "feature",
"priority": 1,
"completed": true,
"tested": true,
"dependencies": [],
"files": [
{
"path": "src/sba_scout/api/importer.py",
"lines": [1, 400],
"issue": "CSV import functions"
}
],
"suggestedFix": null,
"estimatedHours": 4,
"notes": "Completed - imports vL/vR/Total ratings directly from spreadsheet"
},
{
"id": "FEAT-004",
"name": "API Sync",
"description": "Sync players, teams, and transactions from Major Domo league API",
"category": "feature",
"priority": 2,
"completed": true,
"tested": true,
"dependencies": [],
"files": [
{
"path": "src/sba_scout/api/client.py",
"lines": [1, 150],
"issue": "API client implementation"
},
{
"path": "src/sba_scout/api/sync.py",
"lines": [1, 200],
"issue": "Sync logic"
}
],
"suggestedFix": null,
"estimatedHours": 4,
"notes": "Completed - syncs from league API on demand"
},
{
"id": "FEAT-005",
"name": "Matchup Scout Screen",
"description": "Core scouting feature - analyze batter vs pitcher matchups, show expected outcomes based on handedness splits",
"category": "feature",
"priority": 3,
"completed": true,
"tested": true,
"dependencies": ["FEAT-002", "FEAT-003"],
"files": [
{
"path": "src/sba_scout/screens/matchup.py",
"lines": [1, 290],
"issue": "Full implementation complete"
},
{
"path": "src/sba_scout/calc/matchup.py",
"lines": [1, 130],
"issue": "Matchup calculation logic"
}
],
"suggestedFix": null,
"estimatedHours": 8,
"notes": "Implemented: team/pitcher selectors, matchup table with rating/tier, sort by rating/name/position, switch hitter logic (uses opposite of pitcher hand)"
},
{
"id": "FEAT-006",
"name": "Lineup Builder Screen",
"description": "Set batting order and defensive positions, save/load lineup configurations",
"category": "feature",
"priority": 4,
"completed": true,
"tested": true,
"dependencies": ["FEAT-002"],
"files": [
{
"path": "src/sba_scout/screens/lineup.py",
"lines": [1, 450],
"issue": "Full implementation complete"
}
],
"suggestedFix": null,
"estimatedHours": 6,
"notes": "Implemented: dual-panel UI with available batters and lineup tables, keyboard navigation (a=add, r=remove, k/j=move up/down, p=cycle position), save/load named lineups, position eligibility suggestions"
},
{
"id": "FEAT-007",
"name": "Transactions Screen",
"description": "View recent transactions, track player movements, manage roster moves",
"category": "feature",
"priority": 5,
"completed": false,
"tested": false,
"dependencies": ["FEAT-004"],
"files": [
{
"path": "src/sba_scout/app.py",
"lines": [234, 249],
"issue": "Placeholder TransactionsScreen needs implementation"
}
],
"suggestedFix": "1. Create src/sba_scout/screens/transactions.py\n2. Show recent league transactions\n3. Filter by team, player, date\n4. Show demotion weeks for minors players\n5. Highlight players who can be promoted",
"estimatedHours": 4,
"notes": "Transaction model already exists, needs UI"
},
{
"id": "FEAT-008",
"name": "Depth Chart Screen (by Position)",
"description": "View roster organized by defensive position showing depth at each spot",
"category": "feature",
"priority": 6,
"completed": false,
"tested": false,
"dependencies": ["FEAT-002"],
"files": [],
"suggestedFix": "1. Create src/sba_scout/screens/depth_chart.py\n2. Show positions as columns or tabs (C, 1B, 2B, SS, 3B, LF, CF, RF, DH)\n3. List eligible players at each position\n4. Show defensive ratings (range, error, arm)\n5. Highlight primary vs secondary positions",
"estimatedHours": 4,
"notes": "Different from Roster screen - organized by position rather than roster status"
},
{
"id": "FEAT-009",
"name": "Player Detail Modal",
"description": "Click on a player row to see full card details, stats, and ratings breakdown",
"category": "feature",
"priority": 7,
"completed": false,
"tested": false,
"dependencies": ["FEAT-002"],
"files": [
{
"path": "src/sba_scout/screens/roster.py",
"lines": [],
"issue": "Row selection currently does nothing"
}
],
"suggestedFix": "1. Create PlayerDetailModal widget\n2. Show full batter card (all vs LHP and vs RHP stats)\n3. Show full pitcher card (all vs LHB and vs RHB stats)\n4. Show fielding details at each position\n5. Show running game stats (stealing, speed)\n6. Display card image if available",
"estimatedHours": 4,
"notes": "Should work from both Roster and Matchup screens"
},
{
"id": "HIGH-001",
"name": "Dynamic Season Configuration",
"description": "Remove hardcoded season 13, get current season from config or API",
"category": "high",
"priority": 8,
"completed": true,
"tested": true,
"dependencies": [],
"files": [
{
"path": "src/sba_scout/app.py",
"lines": [116],
"issue": "Hardcoded season=13 in dashboard"
},
{
"path": "src/sba_scout/screens/roster.py",
"lines": [138],
"issue": "Hardcoded season=13 in roster"
},
{
"path": "src/sba_scout/api/client.py",
"lines": [32],
"issue": "Example uses season=13"
}
],
"suggestedFix": "1. Add current_season to TeamSettings in config.py\n2. Update .env.example with SBA_SCOUT_TEAM__CURRENT_SEASON=13\n3. Replace all hardcoded 13 with settings.team.current_season\n4. Optionally: fetch current season from league API",
"estimatedHours": 1,
"notes": "Quick fix but important for multi-season support"
},
{
"id": "MED-001",
"name": "Schedule Integration",
"description": "Show upcoming games, opponent info, and starting pitchers",
"category": "medium",
"priority": 9,
"completed": false,
"tested": false,
"dependencies": ["FEAT-004"],
"files": [],
"suggestedFix": "1. Add schedule endpoint to API client\n2. Create Schedule model in database\n3. Add upcoming games widget to dashboard\n4. Link to Matchup Scout with opponent pre-selected",
"estimatedHours": 4,
"notes": "Would enhance dashboard and matchup workflow"
},
{
"id": "MED-002",
"name": "Rating Recalculation Engine",
"description": "Rebuild rating calculation system for matchup analysis (was removed, needs fresh implementation)",
"category": "medium",
"priority": 10,
"completed": false,
"tested": false,
"dependencies": ["FEAT-005"],
"files": [
{
"path": "src/sba_scout/calc/__init__.py",
"lines": [],
"issue": "calc module is empty after removing ratings.py and matchup.py"
}
],
"suggestedFix": "1. Design new calculation approach based on matchup needs\n2. Consider: should matchups combine batter vL + pitcher vRHB?\n3. Implement matchup_score = f(batter_rating, pitcher_rating, handedness)\n4. Add tier system (A/B/C/D/F) for quick reference",
"estimatedHours": 4,
"notes": "Original calc was removed because roster ratings come pre-calculated from spreadsheet. Matchup calculations need different logic."
},
{
"id": "MED-003",
"name": "Keyboard Navigation Enhancement",
"description": "Add vim-style navigation, better hotkeys, command palette",
"category": "medium",
"priority": 11,
"completed": false,
"tested": false,
"dependencies": [],
"files": [],
"suggestedFix": "1. Add j/k navigation in tables\n2. Add search/filter with / key\n3. Implement command palette (ctrl+p)\n4. Add goto player by name",
"estimatedHours": 2,
"notes": "TUI apps benefit greatly from keyboard-first design"
},
{
"id": "MED-004",
"name": "Auto-Sync on Startup",
"description": "Optionally sync data from API when app starts, with configurable interval",
"category": "medium",
"priority": 12,
"completed": false,
"tested": false,
"dependencies": ["FEAT-004"],
"files": [
{
"path": "src/sba_scout/config.py",
"lines": [126, 129],
"issue": "refresh_interval setting exists but not implemented"
}
],
"suggestedFix": "1. Check last_sync timestamp on startup\n2. If stale (> refresh_interval), trigger sync\n3. Show sync status in dashboard\n4. Add background sync timer (optional)",
"estimatedHours": 2,
"notes": "refresh_interval setting already exists in config"
},
{
"id": "LOW-001",
"name": "Theme Support",
"description": "Light/dark theme toggle using Textual's theming system",
"category": "low",
"priority": 13,
"completed": false,
"tested": false,
"dependencies": [],
"files": [
{
"path": "src/sba_scout/config.py",
"lines": [129],
"issue": "theme setting exists but not implemented"
}
],
"suggestedFix": "1. Use Textual's built-in theme system\n2. Add theme toggle hotkey (t)\n3. Persist theme preference in config",
"estimatedHours": 1,
"notes": "theme setting already exists in config"
},
{
"id": "LOW-002",
"name": "Export Functionality",
"description": "Export lineups, matchup reports, or roster data to various formats",
"category": "low",
"priority": 14,
"completed": false,
"tested": false,
"dependencies": ["FEAT-005", "FEAT-006"],
"files": [],
"suggestedFix": "1. Export lineup to clipboard (for pasting into game)\n2. Export matchup report to markdown/text\n3. Export roster to CSV",
"estimatedHours": 2,
"notes": "Clipboard export would be most immediately useful"
},
{
"id": "LOW-003",
"name": "Draft Tools",
"description": "Tools for evaluating players during the draft",
"category": "low",
"priority": 15,
"completed": false,
"tested": false,
"dependencies": ["FEAT-003"],
"files": [],
"suggestedFix": "1. Player search across all teams\n2. Compare players side-by-side\n3. Track draft queue/watchlist\n4. sWAR value analysis",
"estimatedHours": 6,
"notes": "Would be valuable during draft season"
},
{
"id": "TEST-001",
"name": "Add Unit Tests",
"description": "Create test suite for core functionality",
"category": "medium",
"priority": 16,
"completed": false,
"tested": false,
"dependencies": [],
"files": [
{
"path": "tests/__init__.py",
"lines": [],
"issue": "Tests directory is empty"
}
],
"suggestedFix": "1. Test importer with sample CSV data\n2. Test database queries\n3. Test API client with mocked responses\n4. Test config loading",
"estimatedHours": 4,
"notes": "Focus on data layer tests first"
}
],
"quickWins": [
{
"taskId": "HIGH-001",
"estimatedMinutes": 30,
"impact": "Removes hardcoded season number, enables multi-season support"
},
{
"taskId": "LOW-001",
"estimatedMinutes": 30,
"impact": "Light theme option for users who prefer it"
}
],
"productionBlockers": [],
"weeklyRoadmap": {
"week1": {
"theme": "Core Scouting Features",
"tasks": ["HIGH-001", "FEAT-005"],
"estimatedHours": 9,
"description": "Fix hardcoded season, implement Matchup Scout screen"
},
"week2": {
"theme": "Lineup Management",
"tasks": ["FEAT-006", "FEAT-009"],
"estimatedHours": 10,
"description": "Lineup Builder screen and Player Detail modal"
},
"week3": {
"theme": "Secondary Features",
"tasks": ["FEAT-007", "FEAT-008", "MED-002"],
"estimatedHours": 12,
"description": "Transactions, Depth Chart, and matchup calculations"
},
"week4": {
"theme": "Polish and Testing",
"tasks": ["MED-003", "MED-004", "TEST-001"],
"estimatedHours": 8,
"description": "Keyboard navigation, auto-sync, and unit tests"
}
},
"completedHistory": [
{
"taskId": "FEAT-001",
"completedDate": "2026-01-25",
"notes": "Dashboard with roster summary and navigation"
},
{
"taskId": "FEAT-002",
"completedDate": "2026-01-25",
"notes": "Roster screen with batters/pitchers tables, proper ratings from CSV"
},
{
"taskId": "FEAT-003",
"completedDate": "2026-01-25",
"notes": "CSV importer using pre-calculated vL/vR/Total from spreadsheet"
},
{
"taskId": "FEAT-004",
"completedDate": "2026-01-25",
"notes": "API sync for players and teams from Major Domo"
},
{
"taskId": "HIGH-001",
"completedDate": "2026-01-25",
"notes": "Added current_season to TeamSettings config; replaced hardcoded season=13 in app.py and roster.py"
},
{
"taskId": "FEAT-005",
"completedDate": "2026-01-25",
"notes": "Matchup Scout screen with team/pitcher selection, matchup table showing batter ratings vs pitcher handedness, sorting options, tier grades (A-F)"
},
{
"taskId": "FEAT-006",
"completedDate": "2026-01-25",
"notes": "Lineup Builder screen with dual-panel UI (available batters + lineup), keyboard controls (a=add, r=remove, k/j=reorder, p=change position), save/load named lineups"
}
]
}

10
src/sba_scout/api/importer.py
View File

@ -379,4 +379,14 @@ async def import_all_cards(
logger.warning(f"Pitcher CSV not found: {pitcher_csv}") logger.warning(f"Pitcher CSV not found: {pitcher_csv}")
results["pitchers"]["errors"].append(f"File not found: {pitcher_csv}") results["pitchers"]["errors"].append(f"File not found: {pitcher_csv}")
# Rebuild standardized score cache after importing cards
total_imported = results["batters"]["imported"] + results["pitchers"]["imported"]
if total_imported > 0:
from ..calc.score_cache import rebuild_score_cache
logger.info("Rebuilding standardized score cache after import...")
cache_result = await rebuild_score_cache(session)
results["score_cache"] = cache_result
logger.info(f"Score cache rebuilt: {cache_result}")
return results return results

232
src/sba_scout/app.py
View File

@ -17,6 +17,9 @@ from textual.widgets import Button, Footer, Header, Label, LoadingIndicator, Sta
from .config import get_settings from .config import get_settings
from .db.schema import close_database, get_session, init_database from .db.schema import close_database, get_session, init_database
from .screens.gameday import GamedayScreen
from .screens.lineup import LineupScreen
from .screens.matchup import MatchupScreen
from .screens.roster import RosterScreen from .screens.roster import RosterScreen
# Configure logging # Configure logging
@ -33,6 +36,7 @@ class DashboardScreen(Screen):
BINDINGS: ClassVar = [ BINDINGS: ClassVar = [
Binding("r", "switch_screen('roster')", "Roster"), Binding("r", "switch_screen('roster')", "Roster"),
Binding("m", "switch_screen('matchup')", "Matchup Scout"), Binding("m", "switch_screen('matchup')", "Matchup Scout"),
Binding("g", "switch_screen('gameday')", "Gameday"),
Binding("l", "switch_screen('lineup')", "Lineup Builder"), Binding("l", "switch_screen('lineup')", "Lineup Builder"),
Binding("t", "switch_screen('transactions')", "Transactions"), Binding("t", "switch_screen('transactions')", "Transactions"),
Binding("s", "sync_data", "Sync Data"), Binding("s", "sync_data", "Sync Data"),
@ -84,12 +88,12 @@ class DashboardScreen(Screen):
yield Label("Quick Actions", classes="section-title") yield Label("Quick Actions", classes="section-title")
with Horizontal(classes="action-buttons"): with Horizontal(classes="action-buttons"):
yield Button("Gameday [g]", id="btn-gameday", variant="primary")
yield Button("Roster [r]", id="btn-roster", variant="primary") yield Button("Roster [r]", id="btn-roster", variant="primary")
yield Button("Matchup Scout [m]", id="btn-matchup", variant="primary")
with Horizontal(classes="action-buttons"): with Horizontal(classes="action-buttons"):
yield Button("Matchup Scout [m]", id="btn-matchup", variant="default")
yield Button("Lineup Builder [l]", id="btn-lineup", variant="default") yield Button("Lineup Builder [l]", id="btn-lineup", variant="default")
yield Button("Transactions [t]", id="btn-transactions", variant="default")
# Status bar # Status bar
with Horizontal(id="status-bar"): with Horizontal(id="status-bar"):
@ -113,7 +117,7 @@ class DashboardScreen(Screen):
roster = await get_my_roster( roster = await get_my_roster(
session, session,
settings.team.team_abbrev, settings.team.team_abbrev,
13, # TODO: Get current season from API settings.team.current_season,
) )
majors_count = len(roster.get("majors", [])) majors_count = len(roster.get("majors", []))
@ -142,6 +146,11 @@ class DashboardScreen(Screen):
settings = get_settings() settings = get_settings()
self.query_one("#majors-count", Label).update(f"--/{settings.team.major_league_slots}") self.query_one("#majors-count", Label).update(f"--/{settings.team.major_league_slots}")
@on(Button.Pressed, "#btn-gameday")
def on_gameday(self) -> None:
"""Navigate to gameday screen."""
self.app.push_screen("gameday")
@on(Button.Pressed, "#btn-roster") @on(Button.Pressed, "#btn-roster")
def on_roster(self) -> None: def on_roster(self) -> None:
"""Navigate to roster screen.""" """Navigate to roster screen."""
@ -194,35 +203,7 @@ class DashboardScreen(Screen):
sync_btn.disabled = False sync_btn.disabled = False
# RosterScreen is imported from screens.roster # RosterScreen, MatchupScreen, and LineupScreen are imported from screens module
class MatchupScreen(Screen):
"""Matchup scouting screen for analyzing batters vs pitchers."""
BINDINGS: ClassVar = [
Binding("escape", "app.pop_screen", "Back"),
Binding("q", "app.pop_screen", "Back"),
]
def compose(self) -> ComposeResult:
yield Header()
yield Label("Matchup Scout - Coming Soon", id="placeholder")
yield Footer()
class LineupScreen(Screen):
"""Lineup builder for setting batting order and positions."""
BINDINGS: ClassVar = [
Binding("escape", "app.pop_screen", "Back"),
Binding("q", "app.pop_screen", "Back"),
]
def compose(self) -> ComposeResult:
yield Header()
yield Label("Lineup Builder - Coming Soon", id="placeholder")
yield Footer()
class TransactionsScreen(Screen): class TransactionsScreen(Screen):
@ -392,12 +373,199 @@ class SBAScoutApp(App):
DataTable > .datatable--cursor { DataTable > .datatable--cursor {
background: $accent; background: $accent;
} }
/* Matchup Screen Styles */
#matchup-container {
height: 100%;
padding: 1;
}
#matchup-selectors {
height: 3;
margin-bottom: 1;
}
#matchup-selectors Label {
width: auto;
padding: 0 1;
content-align: center middle;
}
#matchup-selectors Select {
width: 1fr;
max-width: 40;
margin-right: 2;
}
#pitcher-info {
height: 2;
background: $surface;
padding: 0 1;
margin-bottom: 1;
text-style: bold;
color: $warning;
}
#matchup-table-container {
height: 1fr;
}
#matchup-table {
width: 100%;
height: 100%;
}
/* Lineup Builder Screen Styles */
#lineup-container {
height: 1fr;
padding: 1;
}
.lineup-panel {
width: 1fr;
height: 100%;
padding: 0 1;
border: solid $primary;
margin-right: 1;
}
.lineup-panel:last-of-type {
margin-right: 0;
}
.panel-title {
text-style: bold;
background: $surface;
padding: 0 1;
margin-bottom: 1;
}
#available-container, #lineup-scroll-container {
height: 1fr;
}
#available-table, #lineup-table {
width: 100%;
height: 100%;
}
.hint-text {
color: $text-muted;
height: 1;
padding: 0 1;
}
#lineup-controls {
height: 5;
padding: 1;
background: $surface;
dock: bottom;
}
#lineup-controls Label {
width: auto;
padding: 0 1;
height: 3;
content-align: center middle;
}
#lineup-name-input {
width: 30;
height: 3;
margin-right: 1;
}
#lineup-select {
width: 25;
height: 3;
margin-right: 1;
}
#btn-save {
width: auto;
margin-right: 1;
}
#btn-clear {
width: auto;
}
/* Gameday Screen Styles */
#gameday-container {
height: 1fr;
padding: 1;
}
.gameday-panel {
height: 100%;
padding: 0 1;
border: solid $primary;
}
#matchup-panel {
width: 3fr;
margin-right: 1;
}
#lineup-panel {
width: 2fr;
}
#gameday-selectors {
height: 3;
margin-bottom: 1;
}
#gameday-selectors Label {
width: auto;
padding: 0 1;
content-align: center middle;
}
#gameday-selectors Select {
width: 1fr;
margin-right: 1;
}
#gameday-selectors #team-select {
max-width: 12;
}
#gameday-selectors #pitcher-select {
max-width: 30;
}
#matchup-scroll, #lineup-scroll {
height: 1fr;
}
#lineup-save-controls {
height: 3;
margin-bottom: 1;
}
#lineup-save-controls Input {
width: 1fr;
height: 3;
margin-right: 1;
}
#lineup-save-controls Select {
width: 1fr;
height: 3;
margin-right: 1;
}
#lineup-save-controls Button {
width: auto;
}
""" """
SCREENS = { SCREENS = {
"dashboard": DashboardScreen, "dashboard": DashboardScreen,
"roster": RosterScreen, "roster": RosterScreen,
"matchup": MatchupScreen, "matchup": MatchupScreen,
"gameday": GamedayScreen,
"lineup": LineupScreen, "lineup": LineupScreen,
"transactions": TransactionsScreen, "transactions": TransactionsScreen,
} }

263
src/sba_scout/calc/league_stats.py Normal file
View File

@ -0,0 +1,263 @@
"""
League statistics calculation for standardized scoring.
Calculates league-wide averages and standard deviations for batter and pitcher
card stats, which are used to convert raw values into standardized scores (-3 to +3).
"""
import statistics
from dataclasses import dataclass
from typing import Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from ..db.models import BatterCard, PitcherCard
@dataclass
class StatDistribution:
"""Average and standard deviation for a single stat."""
avg: float
stdev: float
def __repr__(self) -> str:
return f"StatDistribution(avg={self.avg:.2f}, stdev={self.stdev:.2f})"
@dataclass
class BatterLeagueStats:
"""League-wide averages and standard deviations for batter card stats."""
# vs Left-Handed Pitchers
so_vlhp: StatDistribution
bb_vlhp: StatDistribution
hit_vlhp: StatDistribution
ob_vlhp: StatDistribution
tb_vlhp: StatDistribution
hr_vlhp: StatDistribution
dp_vlhp: StatDistribution
bphr_vlhp: StatDistribution
bp1b_vlhp: StatDistribution
# vs Right-Handed Pitchers
so_vrhp: StatDistribution
bb_vrhp: StatDistribution
hit_vrhp: StatDistribution
ob_vrhp: StatDistribution
tb_vrhp: StatDistribution
hr_vrhp: StatDistribution
dp_vrhp: StatDistribution
bphr_vrhp: StatDistribution
bp1b_vrhp: StatDistribution
@dataclass
class PitcherLeagueStats:
"""League-wide averages and standard deviations for pitcher card stats."""
# vs Left-Handed Batters
so_vlhb: StatDistribution
bb_vlhb: StatDistribution
hit_vlhb: StatDistribution
ob_vlhb: StatDistribution
tb_vlhb: StatDistribution
hr_vlhb: StatDistribution
dp_vlhb: StatDistribution
bphr_vlhb: StatDistribution
bp1b_vlhb: StatDistribution
# vs Right-Handed Batters
so_vrhb: StatDistribution
bb_vrhb: StatDistribution
hit_vrhb: StatDistribution
ob_vrhb: StatDistribution
tb_vrhb: StatDistribution
hr_vrhb: StatDistribution
dp_vrhb: StatDistribution
bphr_vrhb: StatDistribution
bp1b_vrhb: StatDistribution
def _calc_distribution(values: list[float]) -> StatDistribution:
"""Calculate average and standard deviation for a list of values."""
# Filter out None and zero values for average calculation (matching spreadsheet AVERAGEIF)
non_zero = [v for v in values if v and v > 0]
if len(non_zero) < 2:
# Not enough data - return defaults that will make all scores 0
return StatDistribution(avg=0.0, stdev=1.0)
avg = statistics.mean(non_zero)
# Use all values (including zeros) for stdev calculation
all_values = [v or 0 for v in values]
stdev = statistics.stdev(all_values) if len(all_values) >= 2 else 1.0
# Prevent division by zero
if stdev == 0:
stdev = 1.0
return StatDistribution(avg=avg, stdev=stdev)
async def calculate_batter_league_stats(
session: AsyncSession,
) -> BatterLeagueStats:
"""
Calculate league-wide averages and standard deviations for all batter stats.
Queries all batter cards in the database and computes statistics for each
stat column, separated by vs-LHP and vs-RHP splits.
Args:
session: Database session
Returns:
BatterLeagueStats with avg/stdev for each stat
"""
query = select(BatterCard)
result = await session.execute(query)
cards: Sequence[BatterCard] = result.scalars().all()
if not cards:
# Return default stats if no cards exist
default = StatDistribution(avg=0.0, stdev=1.0)
return BatterLeagueStats(
so_vlhp=default,
bb_vlhp=default,
hit_vlhp=default,
ob_vlhp=default,
tb_vlhp=default,
hr_vlhp=default,
dp_vlhp=default,
bphr_vlhp=default,
bp1b_vlhp=default,
so_vrhp=default,
bb_vrhp=default,
hit_vrhp=default,
ob_vrhp=default,
tb_vrhp=default,
hr_vrhp=default,
dp_vrhp=default,
bphr_vrhp=default,
bp1b_vrhp=default,
)
return BatterLeagueStats(
# vs LHP
so_vlhp=_calc_distribution([c.so_vlhp for c in cards]),
bb_vlhp=_calc_distribution([c.bb_vlhp for c in cards]),
hit_vlhp=_calc_distribution([c.hit_vlhp for c in cards]),
ob_vlhp=_calc_distribution([c.ob_vlhp for c in cards]),
tb_vlhp=_calc_distribution([c.tb_vlhp for c in cards]),
hr_vlhp=_calc_distribution([c.hr_vlhp for c in cards]),
dp_vlhp=_calc_distribution([c.dp_vlhp for c in cards]),
bphr_vlhp=_calc_distribution([c.bphr_vlhp for c in cards]),
bp1b_vlhp=_calc_distribution([c.bp1b_vlhp for c in cards]),
# vs RHP
so_vrhp=_calc_distribution([c.so_vrhp for c in cards]),
bb_vrhp=_calc_distribution([c.bb_vrhp for c in cards]),
hit_vrhp=_calc_distribution([c.hit_vrhp for c in cards]),
ob_vrhp=_calc_distribution([c.ob_vrhp for c in cards]),
tb_vrhp=_calc_distribution([c.tb_vrhp for c in cards]),
hr_vrhp=_calc_distribution([c.hr_vrhp for c in cards]),
dp_vrhp=_calc_distribution([c.dp_vrhp for c in cards]),
bphr_vrhp=_calc_distribution([c.bphr_vrhp for c in cards]),
bp1b_vrhp=_calc_distribution([c.bp1b_vrhp for c in cards]),
)
async def calculate_pitcher_league_stats(
session: AsyncSession,
) -> PitcherLeagueStats:
"""
Calculate league-wide averages and standard deviations for all pitcher stats.
Queries all pitcher cards in the database and computes statistics for each
stat column, separated by vs-LHB and vs-RHB splits.
Args:
session: Database session
Returns:
PitcherLeagueStats with avg/stdev for each stat
"""
query = select(PitcherCard)
result = await session.execute(query)
cards: Sequence[PitcherCard] = result.scalars().all()
if not cards:
# Return default stats if no cards exist
default = StatDistribution(avg=0.0, stdev=1.0)
return PitcherLeagueStats(
so_vlhb=default,
bb_vlhb=default,
hit_vlhb=default,
ob_vlhb=default,
tb_vlhb=default,
hr_vlhb=default,
dp_vlhb=default,
bphr_vlhb=default,
bp1b_vlhb=default,
so_vrhb=default,
bb_vrhb=default,
hit_vrhb=default,
ob_vrhb=default,
tb_vrhb=default,
hr_vrhb=default,
dp_vrhb=default,
bphr_vrhb=default,
bp1b_vrhb=default,
)
return PitcherLeagueStats(
# vs LHB
so_vlhb=_calc_distribution([c.so_vlhb for c in cards]),
bb_vlhb=_calc_distribution([c.bb_vlhb for c in cards]),
hit_vlhb=_calc_distribution([c.hit_vlhb for c in cards]),
ob_vlhb=_calc_distribution([c.ob_vlhb for c in cards]),
tb_vlhb=_calc_distribution([c.tb_vlhb for c in cards]),
hr_vlhb=_calc_distribution([c.hr_vlhb for c in cards]),
dp_vlhb=_calc_distribution([c.dp_vlhb for c in cards]),
bphr_vlhb=_calc_distribution([c.bphr_vlhb for c in cards]),
bp1b_vlhb=_calc_distribution([c.bp1b_vlhb for c in cards]),
# vs RHB
so_vrhb=_calc_distribution([c.so_vrhb for c in cards]),
bb_vrhb=_calc_distribution([c.bb_vrhb for c in cards]),
hit_vrhb=_calc_distribution([c.hit_vrhb for c in cards]),
ob_vrhb=_calc_distribution([c.ob_vrhb for c in cards]),
tb_vrhb=_calc_distribution([c.tb_vrhb for c in cards]),
hr_vrhb=_calc_distribution([c.hr_vrhb for c in cards]),
dp_vrhb=_calc_distribution([c.dp_vrhb for c in cards]),
bphr_vrhb=_calc_distribution([c.bphr_vrhb for c in cards]),
bp1b_vrhb=_calc_distribution([c.bp1b_vrhb for c in cards]),
)
# Cached league stats (computed once per session)
_batter_stats_cache: BatterLeagueStats | None = None
_pitcher_stats_cache: PitcherLeagueStats | None = None
async def get_batter_league_stats(session: AsyncSession) -> BatterLeagueStats:
"""Get cached batter league stats, computing if necessary."""
global _batter_stats_cache
if _batter_stats_cache is None:
_batter_stats_cache = await calculate_batter_league_stats(session)
return _batter_stats_cache
async def get_pitcher_league_stats(session: AsyncSession) -> PitcherLeagueStats:
"""Get cached pitcher league stats, computing if necessary."""
global _pitcher_stats_cache
if _pitcher_stats_cache is None:
_pitcher_stats_cache = await calculate_pitcher_league_stats(session)
return _pitcher_stats_cache
def clear_league_stats_cache() -> None:
"""Clear the cached league stats (call when card data changes)."""
global _batter_stats_cache, _pitcher_stats_cache
_batter_stats_cache = None
_pitcher_stats_cache = None

475
src/sba_scout/calc/matchup.py Normal file
View File

@ -0,0 +1,475 @@
"""
Matchup calculation logic for SBA Scout.
Calculates batter performance ratings against specific pitchers using
standardized scoring based on league averages and standard deviations.
The calculation:
1. Convert each raw stat to a standardized score (-3 to +3) based on
how it compares to the league average using standard deviations
2. Multiply by the stat's weight
3. Sum all weighted scores for batter component
4. Do the same for pitcher component
5. INVERT pitcher score (so pitcher allowing hits = good for batter)
6. Total = Batter Component + Inverted Pitcher Component
Supports two modes:
- Real-time calculation (calculate_matchup): Uses league stats passed in
- Cached calculation (calculate_matchup_cached): Uses pre-computed scores from DB
"""
from dataclasses import dataclass
from typing import Literal
from sqlalchemy.ext.asyncio import AsyncSession
from ..db.models import BatterCard, PitcherCard, Player, StandardizedScoreCache
from .league_stats import (
BatterLeagueStats,
PitcherLeagueStats,
)
from .weights import (
BATTER_WEIGHTS,
PITCHER_WEIGHTS,
calculate_weighted_score,
get_max_matchup_score,
)
@dataclass
class MatchupResult:
"""Result of a batter vs pitcher matchup calculation."""
player: Player
rating: float | None # None if no card data
tier: str # A, B, C, D, F, or "--" if no data
batter_hand: str # L, R, or S (actual batting hand used)
batter_split: str # "vLHP" or "vRHP" - batter's split used
pitcher_split: str # "vLHB" or "vRHB" - pitcher's split used
batter_component: float | None # Batter's contribution to rating
pitcher_component: float | None # Pitcher's contribution (before inversion)
@property
def rating_display(self) -> str:
"""Format rating for display (e.g., '+15', '-3', 'N/A')."""
if self.rating is None:
return "N/A"
sign = "+" if self.rating > 0 else ""
return f"{sign}{self.rating:.0f}"
@property
def split_display(self) -> str:
"""Format splits for display (e.g., 'vR/vL')."""
b = "vL" if self.batter_split == "vLHP" else "vR"
p = "vL" if self.pitcher_split == "vLHB" else "vR"
return f"{b}/{p}"
def get_tier(rating: float | None) -> str:
"""
Assign a letter tier based on matchup rating.
With standardized scoring (-3 to +3 per stat, weighted):
- Max batter score: ~66 (all stats at +3)
- Max pitcher score: ~69 (all stats at +3, then inverted)
- Max combined: ~135
Tier thresholds (roughly based on standard deviation bands):
A: >= 40 (Excellent matchup - significantly above average)
B: 20-39 (Good matchup - above average)
C: -19 to 19 (Average matchup)
D: -39 to -20 (Below average matchup)
F: < -40 (Poor matchup - significantly below average)
"""
if rating is None:
return "--"
if rating >= 40:
return "A"
if rating >= 20:
return "B"
if rating >= -19:
return "C"
if rating >= -39:
return "D"
return "F"
def _calculate_batter_component(
card: BatterCard,
vs_hand: Literal["L", "R"],
league_stats: BatterLeagueStats,
) -> float:
"""
Calculate batter's standardized weighted score vs a pitcher's hand.
Args:
card: Batter's card data
vs_hand: Pitcher's throwing hand ("L" or "R")
league_stats: League averages and standard deviations
Returns:
Batter's total weighted score
"""
total = 0.0
if vs_hand == "L":
# vs Left-Handed Pitchers
total += calculate_weighted_score(card.so_vlhp, league_stats.so_vlhp, BATTER_WEIGHTS["so"])
total += calculate_weighted_score(card.bb_vlhp, league_stats.bb_vlhp, BATTER_WEIGHTS["bb"])
total += calculate_weighted_score(
card.hit_vlhp, league_stats.hit_vlhp, BATTER_WEIGHTS["hit"]
)
total += calculate_weighted_score(card.ob_vlhp, league_stats.ob_vlhp, BATTER_WEIGHTS["ob"])
total += calculate_weighted_score(card.tb_vlhp, league_stats.tb_vlhp, BATTER_WEIGHTS["tb"])
total += calculate_weighted_score(card.hr_vlhp, league_stats.hr_vlhp, BATTER_WEIGHTS["hr"])
total += calculate_weighted_score(
card.bphr_vlhp, league_stats.bphr_vlhp, BATTER_WEIGHTS["bphr"]
)
total += calculate_weighted_score(
card.bp1b_vlhp, league_stats.bp1b_vlhp, BATTER_WEIGHTS["bp1b"]
)
total += calculate_weighted_score(card.dp_vlhp, league_stats.dp_vlhp, BATTER_WEIGHTS["dp"])
else:
# vs Right-Handed Pitchers
total += calculate_weighted_score(card.so_vrhp, league_stats.so_vrhp, BATTER_WEIGHTS["so"])
total += calculate_weighted_score(card.bb_vrhp, league_stats.bb_vrhp, BATTER_WEIGHTS["bb"])
total += calculate_weighted_score(
card.hit_vrhp, league_stats.hit_vrhp, BATTER_WEIGHTS["hit"]
)
total += calculate_weighted_score(card.ob_vrhp, league_stats.ob_vrhp, BATTER_WEIGHTS["ob"])
total += calculate_weighted_score(card.tb_vrhp, league_stats.tb_vrhp, BATTER_WEIGHTS["tb"])
total += calculate_weighted_score(card.hr_vrhp, league_stats.hr_vrhp, BATTER_WEIGHTS["hr"])
total += calculate_weighted_score(
card.bphr_vrhp, league_stats.bphr_vrhp, BATTER_WEIGHTS["bphr"]
)
total += calculate_weighted_score(
card.bp1b_vrhp, league_stats.bp1b_vrhp, BATTER_WEIGHTS["bp1b"]
)
total += calculate_weighted_score(card.dp_vrhp, league_stats.dp_vrhp, BATTER_WEIGHTS["dp"])
return total
def _calculate_pitcher_component(
card: PitcherCard,
vs_hand: Literal["L", "R"],
league_stats: PitcherLeagueStats,
) -> float:
"""
Calculate pitcher's standardized weighted score vs a batter's hand.
Note: This returns the pitcher's score from the PITCHER's perspective
(high = good for pitcher). The caller should INVERT this for matchup calc.
Args:
card: Pitcher's card data
vs_hand: Batter's batting hand ("L" or "R")
league_stats: League averages and standard deviations
Returns:
Pitcher's total weighted score (from pitcher's perspective)
"""
total = 0.0
if vs_hand == "L":
# vs Left-Handed Batters
total += calculate_weighted_score(card.so_vlhb, league_stats.so_vlhb, PITCHER_WEIGHTS["so"])
total += calculate_weighted_score(card.bb_vlhb, league_stats.bb_vlhb, PITCHER_WEIGHTS["bb"])
total += calculate_weighted_score(
card.hit_vlhb, league_stats.hit_vlhb, PITCHER_WEIGHTS["hit"]
)
total += calculate_weighted_score(card.ob_vlhb, league_stats.ob_vlhb, PITCHER_WEIGHTS["ob"])
total += calculate_weighted_score(card.tb_vlhb, league_stats.tb_vlhb, PITCHER_WEIGHTS["tb"])
total += calculate_weighted_score(card.hr_vlhb, league_stats.hr_vlhb, PITCHER_WEIGHTS["hr"])
total += calculate_weighted_score(
card.bphr_vlhb, league_stats.bphr_vlhb, PITCHER_WEIGHTS["bphr"]
)
total += calculate_weighted_score(
card.bp1b_vlhb, league_stats.bp1b_vlhb, PITCHER_WEIGHTS["bp1b"]
)
total += calculate_weighted_score(card.dp_vlhb, league_stats.dp_vlhb, PITCHER_WEIGHTS["dp"])
else:
# vs Right-Handed Batters
total += calculate_weighted_score(card.so_vrhb, league_stats.so_vrhb, PITCHER_WEIGHTS["so"])
total += calculate_weighted_score(card.bb_vrhb, league_stats.bb_vrhb, PITCHER_WEIGHTS["bb"])
total += calculate_weighted_score(
card.hit_vrhb, league_stats.hit_vrhb, PITCHER_WEIGHTS["hit"]
)
total += calculate_weighted_score(card.ob_vrhb, league_stats.ob_vrhb, PITCHER_WEIGHTS["ob"])
total += calculate_weighted_score(card.tb_vrhb, league_stats.tb_vrhb, PITCHER_WEIGHTS["tb"])
total += calculate_weighted_score(card.hr_vrhb, league_stats.hr_vrhb, PITCHER_WEIGHTS["hr"])
total += calculate_weighted_score(
card.bphr_vrhb, league_stats.bphr_vrhb, PITCHER_WEIGHTS["bphr"]
)
total += calculate_weighted_score(
card.bp1b_vrhb, league_stats.bp1b_vrhb, PITCHER_WEIGHTS["bp1b"]
)
total += calculate_weighted_score(card.dp_vrhb, league_stats.dp_vrhb, PITCHER_WEIGHTS["dp"])
return total
def calculate_matchup(
player: Player,
batter_card: BatterCard | None,
pitcher: Player,
pitcher_card: PitcherCard | None,
batter_league_stats: BatterLeagueStats,
pitcher_league_stats: PitcherLeagueStats,
) -> MatchupResult:
"""
Calculate batter's expected performance against a specific pitcher.
Uses standardized scoring:
1. Each stat converted to -3 to +3 based on league avg/stdev
2. Multiplied by weight
3. Summed for batter and pitcher components
4. Pitcher component INVERTED (bad pitcher = good for batter)
5. Total = Batter + (-Pitcher)
Args:
player: The batter Player object
batter_card: The batter's card data (may be None)
pitcher: The pitcher Player object
pitcher_card: The pitcher's card data (may be None)
batter_league_stats: League stats for batters
pitcher_league_stats: League stats for pitchers
Returns:
MatchupResult with combined rating, tier, and component breakdown
"""
batter_hand = player.hand or "R"
pitcher_hand = pitcher.hand or "R"
# Determine batter's effective batting hand (for switch hitters)
if batter_hand == "S":
effective_batting_hand: Literal["L", "R"] = "L" if pitcher_hand == "R" else "R"
else:
effective_batting_hand = "L" if batter_hand == "L" else "R"
# Determine which splits to use
batter_split = "vLHP" if pitcher_hand == "L" else "vRHP"
pitcher_split = "vLHB" if effective_batting_hand == "L" else "vRHB"
# No batter card - return N/A result
if batter_card is None:
return MatchupResult(
player=player,
rating=None,
tier="--",
batter_hand=effective_batting_hand,
batter_split=batter_split,
pitcher_split=pitcher_split,
batter_component=None,
pitcher_component=None,
)
# Calculate batter's component
batter_component = _calculate_batter_component(batter_card, pitcher_hand, batter_league_stats)
# Calculate pitcher's component (if available)
if pitcher_card is not None:
pitcher_component = _calculate_pitcher_component(
pitcher_card, effective_batting_hand, pitcher_league_stats
)
# INVERT pitcher component: good pitcher (high score) = bad for batter
total_rating = batter_component + (-pitcher_component)
else:
pitcher_component = None
total_rating = batter_component
return MatchupResult(
player=player,
rating=total_rating,
tier=get_tier(total_rating),
batter_hand=effective_batting_hand,
batter_split=batter_split,
pitcher_split=pitcher_split,
batter_component=batter_component,
pitcher_component=pitcher_component,
)
def calculate_team_matchups(
batters: list[Player],
pitcher: Player,
pitcher_card: PitcherCard | None,
batter_league_stats: BatterLeagueStats,
pitcher_league_stats: PitcherLeagueStats,
) -> list[MatchupResult]:
"""
Calculate matchups for all batters against a specific pitcher.
Args:
batters: List of batter Player objects (with batter_card relationship loaded)
pitcher: The opposing pitcher Player object
pitcher_card: The pitcher's card data (may be None)
batter_league_stats: League stats for batters
pitcher_league_stats: League stats for pitchers
Returns:
List of MatchupResults sorted by rating (highest first), with None ratings last
"""
results = []
for player in batters:
result = calculate_matchup(
player,
player.batter_card,
pitcher,
pitcher_card,
batter_league_stats,
pitcher_league_stats,
)
results.append(result)
# Sort: highest rating first, None ratings at the end
def sort_key(r: MatchupResult) -> tuple[int, float]:
if r.rating is None:
return (1, 0)
return (0, -r.rating)
results.sort(key=sort_key)
return results
# =============================================================================
# Cached Calculation Functions (use pre-computed scores from database)
# =============================================================================
async def calculate_matchup_cached(
session: AsyncSession,
player: Player,
batter_card: BatterCard | None,
pitcher: Player,
pitcher_card: PitcherCard | None,
) -> MatchupResult:
"""
Calculate matchup using pre-computed cached scores from the database.
This is much faster than real-time calculation as it only requires
two simple database lookups instead of computing standardized scores.
Args:
session: Database session for cache lookups
player: The batter Player object
batter_card: The batter's card data (may be None)
pitcher: The pitcher Player object
pitcher_card: The pitcher's card data (may be None)
Returns:
MatchupResult with combined rating, tier, and component breakdown
"""
from .score_cache import get_cached_batter_score, get_cached_pitcher_score
batter_hand = player.hand or "R"
pitcher_hand = pitcher.hand or "R"
# Determine batter's effective batting hand (for switch hitters)
if batter_hand == "S":
effective_batting_hand: Literal["L", "R"] = "L" if pitcher_hand == "R" else "R"
else:
effective_batting_hand = "L" if batter_hand == "L" else "R"
# Determine which splits to use
batter_split_name = "vLHP" if pitcher_hand == "L" else "vRHP"
pitcher_split_name = "vLHB" if effective_batting_hand == "L" else "vRHB"
# Convert to cache key format (lowercase)
batter_split_key = "vlhp" if pitcher_hand == "L" else "vrhp"
pitcher_split_key = "vlhb" if effective_batting_hand == "L" else "vrhb"
# No batter card - return N/A result
if batter_card is None:
return MatchupResult(
player=player,
rating=None,
tier="--",
batter_hand=effective_batting_hand,
batter_split=batter_split_name,
pitcher_split=pitcher_split_name,
batter_component=None,
pitcher_component=None,
)
# Get cached batter score
batter_cache = await get_cached_batter_score(session, batter_card.id, batter_split_key)
if batter_cache is None:
# Cache miss - should not happen if cache is properly maintained
# Fall back to returning None
return MatchupResult(
player=player,
rating=None,
tier="--",
batter_hand=effective_batting_hand,
batter_split=batter_split_name,
pitcher_split=pitcher_split_name,
batter_component=None,
pitcher_component=None,
)
batter_component = batter_cache.total_score
# Get cached pitcher score (if available)
if pitcher_card is not None:
pitcher_cache = await get_cached_pitcher_score(session, pitcher_card.id, pitcher_split_key)
if pitcher_cache is not None:
pitcher_component = pitcher_cache.total_score
# INVERT pitcher component
total_rating = batter_component + (-pitcher_component)
else:
pitcher_component = None
total_rating = batter_component
else:
pitcher_component = None
total_rating = batter_component
return MatchupResult(
player=player,
rating=total_rating,
tier=get_tier(total_rating),
batter_hand=effective_batting_hand,
batter_split=batter_split_name,
pitcher_split=pitcher_split_name,
batter_component=batter_component,
pitcher_component=pitcher_component,
)
async def calculate_team_matchups_cached(
session: AsyncSession,
batters: list[Player],
pitcher: Player,
pitcher_card: PitcherCard | None,
) -> list[MatchupResult]:
"""
Calculate matchups for all batters using cached scores.
Args:
session: Database session for cache lookups
batters: List of batter Player objects (with batter_card relationship loaded)
pitcher: The opposing pitcher Player object
pitcher_card: The pitcher's card data (may be None)
Returns:
List of MatchupResults sorted by rating (highest first), with None ratings last
"""
results = []
for player in batters:
result = await calculate_matchup_cached(
session,
player,
player.batter_card,
pitcher,
pitcher_card,
)
results.append(result)
# Sort: highest rating first, None ratings at the end
def sort_key(r: MatchupResult) -> tuple[int, float]:
if r.rating is None:
return (1, 0)
return (0, -r.rating)
results.sort(key=sort_key)
return results

318
src/sba_scout/calc/score_cache.py Normal file
View File

@ -0,0 +1,318 @@
"""
Standardized score caching system.
Pre-calculates and stores standardized scores for all card stats to avoid
expensive real-time calculations during matchup analysis.
"""
import hashlib
import json
import logging
from dataclasses import asdict, dataclass
from datetime import datetime
from typing import Sequence
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from ..db.models import BatterCard, PitcherCard, StandardizedScoreCache
from .league_stats import (
BatterLeagueStats,
PitcherLeagueStats,
StatDistribution,
calculate_batter_league_stats,
calculate_pitcher_league_stats,
)
from .weights import (
BATTER_WEIGHTS,
PITCHER_WEIGHTS,
StatWeight,
standardize_value,
)
logger = logging.getLogger(__name__)
@dataclass
class StatScore:
"""Score details for a single stat."""
raw: float
std: int # Standardized score (-3 to +3)
weighted: float # std * weight
def _compute_weights_hash() -> str:
"""Generate hash of current weight configuration."""
weights_data = {
"batter": {k: (v.weight, v.high_is_better) for k, v in BATTER_WEIGHTS.items()},
"pitcher": {k: (v.weight, v.high_is_better) for k, v in PITCHER_WEIGHTS.items()},
}
return hashlib.sha256(json.dumps(weights_data, sort_keys=True).encode()).hexdigest()[:16]
def _compute_league_stats_hash(
batter_stats: BatterLeagueStats,
pitcher_stats: PitcherLeagueStats,
) -> str:
"""Generate hash of league statistics."""
# Just hash a few key values to detect significant changes
key_values = [
batter_stats.hit_vrhp.avg,
batter_stats.hit_vrhp.stdev,
batter_stats.so_vrhp.avg,
pitcher_stats.hit_vrhb.avg,
pitcher_stats.so_vrhb.avg,
]
return hashlib.sha256(str(key_values).encode()).hexdigest()[:16]
def _calculate_batter_split_scores(
card: BatterCard,
split: str, # "vlhp" or "vrhp"
league_stats: BatterLeagueStats,
) -> tuple[float, dict[str, dict]]:
"""
Calculate standardized scores for all stats on a batter card split.
Returns:
Tuple of (total_score, stat_scores_dict)
"""
if split == "vlhp":
stats = [
("so", card.so_vlhp, league_stats.so_vlhp),
("bb", card.bb_vlhp, league_stats.bb_vlhp),
("hit", card.hit_vlhp, league_stats.hit_vlhp),
("ob", card.ob_vlhp, league_stats.ob_vlhp),
("tb", card.tb_vlhp, league_stats.tb_vlhp),
("hr", card.hr_vlhp, league_stats.hr_vlhp),
("bphr", card.bphr_vlhp, league_stats.bphr_vlhp),
("bp1b", card.bp1b_vlhp, league_stats.bp1b_vlhp),
("dp", card.dp_vlhp, league_stats.dp_vlhp),
]
else: # vrhp
stats = [
("so", card.so_vrhp, league_stats.so_vrhp),
("bb", card.bb_vrhp, league_stats.bb_vrhp),
("hit", card.hit_vrhp, league_stats.hit_vrhp),
("ob", card.ob_vrhp, league_stats.ob_vrhp),
("tb", card.tb_vrhp, league_stats.tb_vrhp),
("hr", card.hr_vrhp, league_stats.hr_vrhp),
("bphr", card.bphr_vrhp, league_stats.bphr_vrhp),
("bp1b", card.bp1b_vrhp, league_stats.bp1b_vrhp),
("dp", card.dp_vrhp, league_stats.dp_vrhp),
]
total = 0.0
stat_scores = {}
for stat_name, raw_val, dist in stats:
sw = BATTER_WEIGHTS[stat_name]
std_score = standardize_value(raw_val, dist, sw.high_is_better)
weighted = std_score * sw.weight
total += weighted
stat_scores[stat_name] = {
"raw": raw_val or 0,
"std": std_score,
"weighted": weighted,
}
return total, stat_scores
def _calculate_pitcher_split_scores(
card: PitcherCard,
split: str, # "vlhb" or "vrhb"
league_stats: PitcherLeagueStats,
) -> tuple[float, dict[str, dict]]:
"""
Calculate standardized scores for all stats on a pitcher card split.
Returns:
Tuple of (total_score, stat_scores_dict)
"""
if split == "vlhb":
stats = [
("so", card.so_vlhb, league_stats.so_vlhb),
("bb", card.bb_vlhb, league_stats.bb_vlhb),
("hit", card.hit_vlhb, league_stats.hit_vlhb),
("ob", card.ob_vlhb, league_stats.ob_vlhb),
("tb", card.tb_vlhb, league_stats.tb_vlhb),
("hr", card.hr_vlhb, league_stats.hr_vlhb),
("bphr", card.bphr_vlhb, league_stats.bphr_vlhb),
("bp1b", card.bp1b_vlhb, league_stats.bp1b_vlhb),
("dp", card.dp_vlhb, league_stats.dp_vlhb),
]
else: # vrhb
stats = [
("so", card.so_vrhb, league_stats.so_vrhb),
("bb", card.bb_vrhb, league_stats.bb_vrhb),
("hit", card.hit_vrhb, league_stats.hit_vrhb),
("ob", card.ob_vrhb, league_stats.ob_vrhb),
("tb", card.tb_vrhb, league_stats.tb_vrhb),
("hr", card.hr_vrhb, league_stats.hr_vrhb),
("bphr", card.bphr_vrhb, league_stats.bphr_vrhb),
("bp1b", card.bp1b_vrhb, league_stats.bp1b_vrhb),
("dp", card.dp_vrhb, league_stats.dp_vrhb),
]
total = 0.0
stat_scores = {}
for stat_name, raw_val, dist in stats:
sw = PITCHER_WEIGHTS[stat_name]
std_score = standardize_value(raw_val, dist, sw.high_is_better)
weighted = std_score * sw.weight
total += weighted
stat_scores[stat_name] = {
"raw": raw_val or 0,
"std": std_score,
"weighted": weighted,
}
return total, stat_scores
async def rebuild_score_cache(session: AsyncSession) -> dict[str, int]:
"""
Rebuild the entire standardized score cache.
Deletes all existing cached scores and recalculates from scratch
using current league statistics and weights.
Args:
session: Database session
Returns:
Dict with counts: {"batter_splits": N, "pitcher_splits": M}
"""
logger.info("Rebuilding standardized score cache...")
# Calculate current league stats
batter_league_stats = await calculate_batter_league_stats(session)
pitcher_league_stats = await calculate_pitcher_league_stats(session)
# Get hashes for cache validity tracking
weights_hash = _compute_weights_hash()
league_hash = _compute_league_stats_hash(batter_league_stats, pitcher_league_stats)
# Clear existing cache
await session.execute(delete(StandardizedScoreCache))
# Get all batter cards
batter_result = await session.execute(select(BatterCard))
batter_cards: Sequence[BatterCard] = batter_result.scalars().all()
# Get all pitcher cards
pitcher_result = await session.execute(select(PitcherCard))
pitcher_cards: Sequence[PitcherCard] = pitcher_result.scalars().all()
batter_count = 0
pitcher_count = 0
now = datetime.utcnow()
# Calculate and cache batter scores
for card in batter_cards:
for split in ["vlhp", "vrhp"]:
total, stat_scores = _calculate_batter_split_scores(card, split, batter_league_stats)
cache_entry = StandardizedScoreCache(
batter_card_id=card.id,
pitcher_card_id=None,
split=split,
total_score=total,
stat_scores=stat_scores,
computed_at=now,
weights_hash=weights_hash,
league_stats_hash=league_hash,
)
session.add(cache_entry)
batter_count += 1
# Calculate and cache pitcher scores
for card in pitcher_cards:
for split in ["vlhb", "vrhb"]:
total, stat_scores = _calculate_pitcher_split_scores(card, split, pitcher_league_stats)
cache_entry = StandardizedScoreCache(
batter_card_id=None,
pitcher_card_id=card.id,
split=split,
total_score=total,
stat_scores=stat_scores,
computed_at=now,
weights_hash=weights_hash,
league_stats_hash=league_hash,
)
session.add(cache_entry)
pitcher_count += 1
await session.flush()
logger.info(
f"Score cache rebuilt: {batter_count} batter splits, {pitcher_count} pitcher splits"
)
return {"batter_splits": batter_count, "pitcher_splits": pitcher_count}
async def get_cached_batter_score(
session: AsyncSession,
batter_card_id: int,
split: str, # "vlhp" or "vrhp"
) -> StandardizedScoreCache | None:
"""Get cached score for a batter card split."""
query = select(StandardizedScoreCache).where(
StandardizedScoreCache.batter_card_id == batter_card_id,
StandardizedScoreCache.split == split,
)
result = await session.execute(query)
return result.scalar_one_or_none()
async def get_cached_pitcher_score(
session: AsyncSession,
pitcher_card_id: int,
split: str, # "vlhb" or "vrhb"
) -> StandardizedScoreCache | None:
"""Get cached score for a pitcher card split."""
query = select(StandardizedScoreCache).where(
StandardizedScoreCache.pitcher_card_id == pitcher_card_id,
StandardizedScoreCache.split == split,
)
result = await session.execute(query)
return result.scalar_one_or_none()
async def is_cache_valid(session: AsyncSession) -> bool:
"""
Check if the score cache is valid (exists and has current weights/stats).
Returns False if:
- Cache is empty
- Weights have changed
- League stats have changed significantly
"""
# Check if any cache entries exist
query = select(StandardizedScoreCache).limit(1)
result = await session.execute(query)
entry = result.scalar_one_or_none()
if entry is None:
return False
# Check weights hash
current_weights_hash = _compute_weights_hash()
if entry.weights_hash != current_weights_hash:
logger.info("Cache invalid: weights have changed")
return False
# Could also check league_stats_hash here, but that would require
# recalculating stats which defeats the purpose of caching.
# For now, trust that cache is rebuilt when cards are imported.
return True
async def ensure_cache_exists(session: AsyncSession) -> None:
"""Ensure score cache exists, rebuilding if necessary."""
if not await is_cache_valid(session):
await rebuild_score_cache(session)
await session.commit()

170
src/sba_scout/calc/weights.py Normal file
View File

@ -0,0 +1,170 @@
"""
Stat weights and standardized scoring for matchup calculations.
Converts raw card values into standardized scores (-3 to +3) based on
league averages and standard deviations, then applies weights.
"""
from dataclasses import dataclass
from typing import Literal
from .league_stats import StatDistribution
@dataclass
class StatWeight:
"""Weight and direction for a single stat."""
weight: int
high_is_better: bool # If True, high values get positive scores
# =============================================================================
# Batter Stat Weights (for matchup calculation)
# =============================================================================
BATTER_WEIGHTS: dict[str, StatWeight] = {
"so": StatWeight(weight=1, high_is_better=False), # Strikeouts - low is better
"bb": StatWeight(weight=1, high_is_better=True), # Walks - high is better
"hit": StatWeight(weight=2, high_is_better=True), # Hits - high is better
"ob": StatWeight(weight=5, high_is_better=True), # On-base - high is better
"tb": StatWeight(weight=5, high_is_better=True), # Total bases - high is better
"hr": StatWeight(weight=2, high_is_better=True), # Home runs - high is better
"bphr": StatWeight(weight=3, high_is_better=True), # Ballpark HR - high is better
"bp1b": StatWeight(weight=1, high_is_better=True), # Ballpark 1B - high is better
"dp": StatWeight(weight=2, high_is_better=False), # Double plays - low is better
}
# =============================================================================
# Pitcher Stat Weights (for matchup calculation)
# =============================================================================
PITCHER_WEIGHTS: dict[str, StatWeight] = {
"so": StatWeight(weight=3, high_is_better=True), # Strikeouts - high is better for pitcher
"bb": StatWeight(weight=1, high_is_better=False), # Walks - low is better for pitcher
"hit": StatWeight(weight=2, high_is_better=False), # Hits - low is better for pitcher
"ob": StatWeight(weight=5, high_is_better=False), # On-base - low is better for pitcher
"tb": StatWeight(weight=2, high_is_better=False), # Total bases - low is better for pitcher
"hr": StatWeight(weight=5, high_is_better=False), # Home runs - low is better for pitcher
"bphr": StatWeight(weight=2, high_is_better=False), # Ballpark HR - low is better for pitcher
"bp1b": StatWeight(weight=1, high_is_better=False), # Ballpark 1B - low is better for pitcher
"dp": StatWeight(weight=2, high_is_better=True), # Double plays - high is better for pitcher
}
# =============================================================================
# Standardized Scoring Functions
# =============================================================================
def standardize_value(
value: float | None,
distribution: StatDistribution,
high_is_better: bool,
) -> int:
"""
Convert a raw stat value to a standardized score (-3 to +3).
Uses the following thresholds based on standard deviations from the mean:
> AVG + 2*STDEV: -3 (or +3 if high_is_better)
> AVG + 1*STDEV: -2 (or +2)
> AVG + 0.33*STDEV: -1 (or +1)
> AVG - 0.33*STDEV: 0
> AVG - 1*STDEV: +1 (or -1)
> AVG - 2*STDEV: +2 (or -2)
else: +3 (or -3)
Special case: value of 0 gets the best score (+3 for low_is_better, +3 for high after invert)
Args:
value: Raw stat value from card
distribution: League average and standard deviation
high_is_better: If True, high values get positive scores (inverted)
Returns:
Standardized score from -3 to +3
"""
if value is None or value == 0:
# Zero value = best possible (for stats like SO, HR where 0 is rare/great)
return 3 if not high_is_better else 3
avg = distribution.avg
stdev = distribution.stdev
# Calculate thresholds
thresh_plus_2sd = avg + (2 * stdev)
thresh_plus_1sd = avg + (1 * stdev)
thresh_plus_033sd = avg + (0.33 * stdev)
thresh_minus_033sd = avg - (0.33 * stdev)
thresh_minus_1sd = avg - (1 * stdev)
thresh_minus_2sd = avg - (2 * stdev)
# Determine base score (before inversion)
# High values get negative scores in base formula
if value > thresh_plus_2sd:
base_score = -3
elif value > thresh_plus_1sd:
base_score = -2
elif value > thresh_plus_033sd:
base_score = -1
elif value > thresh_minus_033sd:
base_score = 0
elif value > thresh_minus_1sd:
base_score = 1
elif value > thresh_minus_2sd:
base_score = 2
else:
base_score = 3
# Invert if high values are better
if high_is_better:
return -base_score
return base_score
def calculate_weighted_score(
value: float | None,
distribution: StatDistribution,
stat_weight: StatWeight,
) -> float:
"""
Calculate weighted score for a single stat.
Args:
value: Raw stat value
distribution: League avg/stdev for this stat
stat_weight: Weight and direction for this stat
Returns:
Weighted score (standardized_score * weight)
"""
std_score = standardize_value(value, distribution, stat_weight.high_is_better)
return std_score * stat_weight.weight
# =============================================================================
# Maximum Possible Scores (for reference)
# =============================================================================
def get_max_batter_score() -> int:
"""Get the maximum possible batter component score."""
# All stats at +3, multiplied by weights
return sum(3 * w.weight for w in BATTER_WEIGHTS.values())
def get_max_pitcher_score() -> int:
"""Get the maximum possible pitcher component score."""
return sum(3 * w.weight for w in PITCHER_WEIGHTS.values())
def get_max_matchup_score() -> int:
"""Get the maximum possible combined matchup score."""
return get_max_batter_score() + get_max_pitcher_score()
# Max scores:
# Batter: (1+1+2+5+5+2+3+1+2) * 3 = 22 * 3 = 66
# Pitcher: (3+1+2+5+2+5+2+1+2) * 3 = 23 * 3 = 69
# Combined max: 135

1
src/sba_scout/config.py
View File

@ -85,6 +85,7 @@ class TeamSettings(BaseModel):
team_id: int = Field(default=548, description="Your team's ID in the league") team_id: int = Field(default=548, description="Your team's ID in the league")
team_abbrev: str = Field(default="WV", description="Your team's abbreviation") team_abbrev: str = Field(default="WV", description="Your team's abbreviation")
team_name: str = Field(default="West Virginia Black Bears", description="Full team name") team_name: str = Field(default="West Virginia Black Bears", description="Full team name")
current_season: int = Field(default=13, description="Current season number")
swar_cap: float = Field(default=29.5, description="Season sWAR cap for roster") swar_cap: float = Field(default=29.5, description="Season sWAR cap for roster")
minor_league_slots: int = Field(default=5, description="Number of minor league roster slots") minor_league_slots: int = Field(default=5, description="Number of minor league roster slots")
major_league_slots: int = Field(default=26, description="Number of major league roster slots") major_league_slots: int = Field(default=26, description="Number of major league roster slots")

46
src/sba_scout/db/models.py
View File

@ -410,6 +410,52 @@ class MatchupCache(Base):
) )
class StandardizedScoreCache(Base):
"""
Cached standardized scores for card stats.
Pre-computes the standardized score (-3 to +3) and weighted score for each
stat on each card, based on league averages and standard deviations.
Invalidated and recalculated when:
- Card data is imported/updated
- Weight values are changed
- League stats change significantly (new cards added)
"""
__tablename__ = "standardized_score_cache"
id = Column(Integer, primary_key=True, autoincrement=True)
# Card reference (either batter or pitcher, not both)
batter_card_id = Column(Integer, ForeignKey("batter_cards.id"), nullable=True)
pitcher_card_id = Column(Integer, ForeignKey("pitcher_cards.id"), nullable=True)
# Which split this score is for
split = Column(String(10), nullable=False) # "vlhp", "vrhp", "vlhb", "vrhb"
# Pre-computed total weighted score for this card/split
total_score = Column(Float, nullable=False)
# Individual stat scores (JSON for flexibility)
# Format: {"so": {"raw": 15.0, "std": 1, "weighted": 1}, "bb": {...}, ...}
stat_scores = Column(JSON, nullable=False)
# Cache validity
computed_at = Column(DateTime, default=datetime.utcnow)
weights_hash = Column(String(64), nullable=True) # Hash of weights used
league_stats_hash = Column(String(64), nullable=True) # Hash of league avg/stdev
# Relationships
batter_card = relationship("BatterCard", foreign_keys=[batter_card_id])
pitcher_card = relationship("PitcherCard", foreign_keys=[pitcher_card_id])
__table_args__ = (
UniqueConstraint("batter_card_id", "split", name="uq_batter_score_split"),
UniqueConstraint("pitcher_card_id", "split", name="uq_pitcher_score_split"),
)
class SyncStatus(Base): class SyncStatus(Base):
""" """
Tracks sync status with the league API. Tracks sync status with the league API.

16
src/sba_scout/db/queries.py
View File

@ -463,3 +463,19 @@ async def save_lineup(
) )
session.add(lineup) session.add(lineup)
return lineup return lineup
async def delete_lineup(
session: AsyncSession,
name: str,
) -> bool:
"""Delete a lineup by name.
Returns:
True if lineup was deleted, False if not found
"""
lineup = await get_lineup_by_name(session, name)
if lineup:
await session.delete(lineup)
return True
return False

691
src/sba_scout/screens/gameday.py Normal file
View File

@ -0,0 +1,691 @@
"""
Gameday Screen - Integrated Matchup Scout + Lineup Builder.
Pre-game workflow: analyze matchups against opposing pitcher,
then build your lineup based on the matchup advantages.
Left panel: Matchup analysis (team/pitcher selection, batter ratings)
Right panel: Lineup builder (batting order with positions)
"""
import logging
from dataclasses import dataclass
from typing import ClassVar, Optional
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, ScrollableContainer, Vertical
from textual.screen import Screen
from textual.widgets import (
Button,
DataTable,
Footer,
Header,
Input,
Label,
Select,
Static,
)
from ..calc.matchup import MatchupResult, calculate_team_matchups_cached
from ..calc.score_cache import ensure_cache_exists
from ..config import get_settings
from ..db.models import Lineup, Player, Team
from ..db.queries import (
delete_lineup,
get_all_teams,
get_lineups,
get_lineup_by_name,
get_my_roster,
get_pitchers,
save_lineup,
)
from ..db.schema import get_session
logger = logging.getLogger(__name__)
# Defensive positions
DEFENSIVE_POSITIONS = ["C", "1B", "2B", "3B", "SS", "LF", "CF", "RF", "DH"]
@dataclass
class LineupSlot:
"""A slot in the batting order with position assignment."""
order: int # 1-9
player: Optional[Player] = None
position: Optional[str] = None
matchup_rating: Optional[float] = None # Rating vs current pitcher
class GamedayScreen(Screen):
"""Integrated matchup scout and lineup builder for game day preparation."""
BINDINGS: ClassVar = [
Binding("escape", "app.pop_screen", "Back"),
Binding("q", "app.pop_screen", "Back"),
# Matchup controls
Binding("s", "sort_by_rating", "Sort: Rating"),
Binding("n", "sort_by_name", "Sort: Name"),
# Lineup controls
Binding("a", "add_to_lineup", "Add to Lineup"),
Binding("r", "remove_from_lineup", "Remove"),
Binding("k", "move_up", "Move Up"),
Binding("j", "move_down", "Move Down"),
Binding("p", "cycle_position", "Change Pos"),
Binding("ctrl+s", "save_lineup", "Save Lineup"),
Binding("c", "clear_lineup", "Clear"),
]
# Matchup state
teams: list[Team] = []
opponent_pitchers: list[Player] = []
my_batters: list[Player] = []
selected_team_id: int | None = None
selected_pitcher: Player | None = None
matchup_results: list[MatchupResult] = []
current_sort: str = "rating"
# Lineup state
lineup_slots: list[LineupSlot] = []
saved_lineups: list[Lineup] = []
current_lineup_name: str = ""
def compose(self) -> ComposeResult:
"""Compose the gameday layout with side-by-side panels."""
yield Header()
with Horizontal(id="gameday-container"):
# Left panel - Matchup Scout
with Vertical(id="matchup-panel", classes="gameday-panel"):
yield Label("Matchup Scout", classes="panel-title")
# Team/Pitcher selectors
with Horizontal(id="gameday-selectors"):
yield Label("vs")
yield Select(
options=[],
prompt="Team",
id="team-select",
allow_blank=True,
)
yield Select(
options=[],
prompt="Pitcher",
id="pitcher-select",
allow_blank=True,
)
# Pitcher info
yield Static("Select opponent", id="pitcher-info")
# Matchup table
with ScrollableContainer(id="matchup-scroll"):
yield DataTable(
id="matchup-table",
cursor_type="row",
zebra_stripes=True,
)
yield Static("[a] Add to lineup [s/n] Sort", classes="hint-text")
# Right panel - Lineup Builder
with Vertical(id="lineup-panel", classes="gameday-panel"):
# Save/Load controls at top
with Horizontal(id="lineup-save-controls"):
yield Input(
placeholder="Lineup name",
id="lineup-name-input",
)
yield Select(
options=[],
prompt="Load",
id="lineup-select",
allow_blank=True,
)
yield Button("Save", id="btn-save", variant="success")
yield Label("Game Lineup", classes="panel-title")
# Lineup table
with ScrollableContainer(id="lineup-scroll"):
yield DataTable(
id="lineup-table",
cursor_type="row",
zebra_stripes=True,
)
yield Static("[k/j] Move [p] Pos [r] Remove", classes="hint-text")
yield Footer()
async def on_mount(self) -> None:
"""Initialize when screen mounts."""
self._init_lineup_slots()
await self._setup_tables()
await self._ensure_score_cache()
await self._load_teams()
await self._load_my_batters()
await self._load_saved_lineups()
def _init_lineup_slots(self) -> None:
"""Initialize empty lineup slots for 9 batters."""
self.lineup_slots = [LineupSlot(order=i) for i in range(1, 10)]
async def _setup_tables(self) -> None:
"""Configure both data tables."""
# Matchup table
matchup_table = self.query_one("#matchup-table", DataTable)
matchup_table.add_column("#", key="rank", width=3)
matchup_table.add_column("Name", key="name", width=18)
matchup_table.add_column("Pos", key="pos", width=12)
matchup_table.add_column("H", key="hand", width=2)
matchup_table.add_column("Rating", key="rating", width=7)
matchup_table.add_column("Tier", key="tier", width=4)
# Lineup table
lineup_table = self.query_one("#lineup-table", DataTable)
lineup_table.add_column("#", key="order", width=2)
lineup_table.add_column("H", key="hand", width=2)
lineup_table.add_column("Name", key="name", width=16)
lineup_table.add_column("Pos", key="pos", width=4)
lineup_table.add_column("Rating", key="rating", width=7)
lineup_table.add_column("Tier", key="tier", width=4)
async def _ensure_score_cache(self) -> None:
"""Ensure standardized score cache exists."""
try:
async with get_session() as session:
await ensure_cache_exists(session)
except Exception as e:
logger.error(f"Failed to ensure score cache: {e}")
async def _load_teams(self) -> None:
"""Load all teams for the opponent selector."""
settings = get_settings()
try:
async with get_session() as session:
self.teams = list(
await get_all_teams(session, settings.team.current_season, active_only=True)
)
team_select = self.query_one("#team-select", Select)
options = [
(f"{t.abbrev}", t.id) for t in self.teams if t.abbrev != settings.team.team_abbrev
]
team_select.set_options(options)
except Exception as e:
logger.error(f"Failed to load teams: {e}")
async def _load_my_batters(self) -> None:
"""Load my team's batters."""
settings = get_settings()
try:
async with get_session() as session:
roster = await get_my_roster(
session,
settings.team.team_abbrev,
settings.team.current_season,
)
self.my_batters = [p for p in roster.get("majors", []) if p.is_batter]
except Exception as e:
logger.error(f"Failed to load batters: {e}")
async def _load_opponent_pitchers(self, team_id: int) -> None:
"""Load pitchers for the selected opponent team."""
settings = get_settings()
try:
async with get_session() as session:
self.opponent_pitchers = list(
await get_pitchers(
session, team_id=team_id, season=settings.team.current_season
)
)
pitcher_select = self.query_one("#pitcher-select", Select)
options = [(f"{p.name} ({p.hand or '?'})", p.id) for p in self.opponent_pitchers]
pitcher_select.set_options(options)
self.selected_pitcher = None
self.query_one("#pitcher-info", Static).update("Select a pitcher")
self._clear_matchup_table()
except Exception as e:
logger.error(f"Failed to load pitchers: {e}")
async def _load_saved_lineups(self) -> None:
"""Load saved lineups for the dropdown."""
try:
async with get_session() as session:
self.saved_lineups = list(await get_lineups(session))
lineup_select = self.query_one("#lineup-select", Select)
options = [(lu.name, lu.name) for lu in self.saved_lineups]
lineup_select.set_options(options)
except Exception as e:
logger.error(f"Failed to load saved lineups: {e}")
def _clear_matchup_table(self) -> None:
"""Clear the matchup table."""
table = self.query_one("#matchup-table", DataTable)
table.clear()
self.matchup_results = []
async def _calculate_and_display_matchups(self) -> None:
"""Calculate matchups and populate the table."""
if not self.selected_pitcher:
return
pitcher_card = self.selected_pitcher.pitcher_card
async with get_session() as session:
self.matchup_results = await calculate_team_matchups_cached(
session,
self.my_batters,
self.selected_pitcher,
pitcher_card,
)
self._apply_sort()
self._populate_matchup_table()
self._update_lineup_ratings()
# Update pitcher info
pitcher_info = self.query_one("#pitcher-info", Static)
hand = self.selected_pitcher.hand or "?"
card_status = "" if pitcher_card else " [no card]"
pitcher_info.update(f"vs {self.selected_pitcher.name} ({hand}HP){card_status}")
def _apply_sort(self) -> None:
"""Sort matchup results."""
if self.current_sort == "rating":
def rating_key(r: MatchupResult) -> tuple[int, float]:
if r.rating is None:
return (1, 0)
return (0, -r.rating)
self.matchup_results.sort(key=rating_key)
elif self.current_sort == "name":
self.matchup_results.sort(key=lambda r: r.player.name)
def _populate_matchup_table(self) -> None:
"""Populate the matchup table."""
table = self.query_one("#matchup-table", DataTable)
table.clear()
# Get players already in lineup
lineup_player_ids = {slot.player.id for slot in self.lineup_slots if slot.player}
for i, result in enumerate(self.matchup_results, start=1):
player = result.player
positions = ", ".join(player.positions[:3]) # Limit to 3 positions
# Mark if already in lineup
name = player.name
if player.id in lineup_player_ids:
name = f"* {name}"
row = (
str(i),
name,
positions,
result.batter_hand,
result.rating_display,
result.tier,
)
table.add_row(*row, key=str(player.id))
def _populate_lineup_table(self) -> None:
"""Populate the lineup table."""
table = self.query_one("#lineup-table", DataTable)
cursor_row = table.cursor_row
table.clear()
# Build rating lookup from current matchup results
rating_lookup = {r.player.id: r for r in self.matchup_results}
for slot in self.lineup_slots:
if slot.player:
# Get matchup rating if available
matchup = rating_lookup.get(slot.player.id)
rating_str = matchup.rating_display if matchup else "---"
tier_str = matchup.tier if matchup else "-"
hand = slot.player.hand or "?"
row = (
str(slot.order),
hand,
slot.player.name,
slot.position or "---",
rating_str,
tier_str,
)
else:
row = (str(slot.order), "-", "---", "---", "---", "-")
table.add_row(*row, key=f"slot-{slot.order}")
# Restore cursor
if cursor_row is not None and cursor_row < 9:
table.move_cursor(row=cursor_row)
def _update_lineup_ratings(self) -> None:
"""Update ratings in lineup table after matchup calculation."""
self._populate_lineup_table()
def _get_selected_matchup_player(self) -> Optional[Player]:
"""Get the currently selected player from matchup table."""
table = self.query_one("#matchup-table", DataTable)
cursor_row = table.cursor_row
if cursor_row is None:
return None
try:
row_keys = list(table.rows.keys())
if cursor_row >= len(row_keys):
return None
player_id = int(row_keys[cursor_row].value)
return next((p for p in self.my_batters if p.id == player_id), None)
except (IndexError, ValueError, AttributeError):
return None
def _get_selected_lineup_slot(self) -> Optional[LineupSlot]:
"""Get the currently selected lineup slot."""
table = self.query_one("#lineup-table", DataTable)
if table.cursor_row is None:
return None
if 0 <= table.cursor_row < len(self.lineup_slots):
return self.lineup_slots[table.cursor_row]
return None
def _find_first_empty_slot(self) -> Optional[LineupSlot]:
"""Find the first empty slot in the lineup."""
for slot in self.lineup_slots:
if slot.player is None:
return slot
return None
def _suggest_position(self, player: Player) -> str:
"""Suggest the best available position for a player."""
positions = player.positions
filled_positions = {
slot.position for slot in self.lineup_slots if slot.position and slot.player
}
for pos in positions:
if pos in DEFENSIVE_POSITIONS and pos not in filled_positions:
return pos
if "DH" not in filled_positions:
return "DH"
return positions[0] if positions else "DH"
def _get_next_position(self, player: Player, current: Optional[str]) -> str:
"""Get next valid position for cycling."""
positions = list(player.positions)
if "DH" not in positions:
positions.append("DH")
if not current or current not in positions:
return positions[0] if positions else "DH"
idx = positions.index(current)
return positions[(idx + 1) % len(positions)]
# =========================================================================
# Event Handlers
# =========================================================================
async def on_select_changed(self, event: Select.Changed) -> None:
"""Handle selection changes."""
if event.select.id == "team-select":
if event.value and event.value != Select.BLANK:
self.selected_team_id = event.value
await self._load_opponent_pitchers(event.value)
else:
self.selected_team_id = None
self.query_one("#pitcher-select", Select).set_options([])
self._clear_matchup_table()
self.query_one("#pitcher-info", Static).update("Select opponent")
elif event.select.id == "pitcher-select":
if event.value and event.value != Select.BLANK:
self.selected_pitcher = next(
(p for p in self.opponent_pitchers if p.id == event.value), None
)
await self._calculate_and_display_matchups()
else:
self.selected_pitcher = None
self._clear_matchup_table()
self.query_one("#pitcher-info", Static).update("Select a pitcher")
elif event.select.id == "lineup-select":
if event.value and event.value != Select.BLANK:
await self._do_load_lineup(event.value)
async def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button clicks."""
if event.button.id == "btn-save":
await self.action_save_lineup()
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
"""Handle row selection - toggle cursor off if clicking same row."""
table = event.data_table
if table.id == "lineup-table":
# If cursor is "row" type, switch to "none" to hide highlight
# User can click again or use arrow keys to restore
if table.cursor_type == "row":
table.cursor_type = "none"
else:
table.cursor_type = "row"
def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
"""Re-enable cursor when navigating with keys."""
table = event.data_table
if table.id == "lineup-table" and table.cursor_type == "none":
# Re-enable cursor when user navigates
table.cursor_type = "row"
# =========================================================================
# Actions
# =========================================================================
def action_sort_by_rating(self) -> None:
"""Sort matchup table by rating."""
self.current_sort = "rating"
if self.matchup_results:
self._apply_sort()
self._populate_matchup_table()
def action_sort_by_name(self) -> None:
"""Sort matchup table by name."""
self.current_sort = "name"
if self.matchup_results:
self._apply_sort()
self._populate_matchup_table()
async def action_add_to_lineup(self) -> None:
"""Add selected matchup batter to lineup."""
player = self._get_selected_matchup_player()
if not player:
self.notify("Select a batter from matchup results", severity="warning")
return
# Check if already in lineup
if any(slot.player and slot.player.id == player.id for slot in self.lineup_slots):
self.notify(f"{player.name} already in lineup", severity="warning")
return
slot = self._find_first_empty_slot()
if not slot:
self.notify("Lineup is full (9 batters)", severity="warning")
return
slot.player = player
slot.position = self._suggest_position(player)
added_slot_idx = slot.order - 1
self._populate_matchup_table() # Update to show * prefix
self._populate_lineup_table()
# Move lineup cursor to new player
lineup_table = self.query_one("#lineup-table", DataTable)
lineup_table.move_cursor(row=added_slot_idx)
self.notify(f"Added {player.name} at {slot.position}")
async def action_remove_from_lineup(self) -> None:
"""Remove selected player from lineup."""
lineup_table = self.query_one("#lineup-table", DataTable)
cursor_row = lineup_table.cursor_row
slot = self._get_selected_lineup_slot()
if not slot or not slot.player:
self.notify("Select a player in the lineup", severity="warning")
return
player_name = slot.player.name
slot.player = None
slot.position = None
self._populate_matchup_table() # Update to remove * prefix
self._populate_lineup_table()
if cursor_row is not None:
lineup_table.move_cursor(row=cursor_row)
self.notify(f"Removed {player_name}")
async def action_move_up(self) -> None:
"""Move selected slot up in batting order."""
slot = self._get_selected_lineup_slot()
if not slot:
return
idx = slot.order - 1
if idx <= 0:
return
self.lineup_slots[idx], self.lineup_slots[idx - 1] = (
self.lineup_slots[idx - 1],
self.lineup_slots[idx],
)
for i, s in enumerate(self.lineup_slots):
s.order = i + 1
self._populate_lineup_table()
self.query_one("#lineup-table", DataTable).move_cursor(row=idx - 1)
async def action_move_down(self) -> None:
"""Move selected slot down in batting order."""
slot = self._get_selected_lineup_slot()
if not slot:
return
idx = slot.order - 1
if idx >= 8:
return
self.lineup_slots[idx], self.lineup_slots[idx + 1] = (
self.lineup_slots[idx + 1],
self.lineup_slots[idx],
)
for i, s in enumerate(self.lineup_slots):
s.order = i + 1
self._populate_lineup_table()
self.query_one("#lineup-table", DataTable).move_cursor(row=idx + 1)
async def action_cycle_position(self) -> None:
"""Cycle through positions for selected player."""
lineup_table = self.query_one("#lineup-table", DataTable)
cursor_row = lineup_table.cursor_row
slot = self._get_selected_lineup_slot()
if not slot or not slot.player:
self.notify("Select a player in the lineup", severity="warning")
return
slot.position = self._get_next_position(slot.player, slot.position)
self._populate_lineup_table()
if cursor_row is not None:
lineup_table.move_cursor(row=cursor_row)
self.notify(f"{slot.player.name} -> {slot.position}")
async def action_save_lineup(self) -> None:
"""Save current lineup."""
name_input = self.query_one("#lineup-name-input", Input)
name = name_input.value.strip()
if not name:
self.notify("Enter a lineup name", severity="warning")
name_input.focus()
return
batting_order = []
positions = {}
for slot in self.lineup_slots:
if slot.player:
batting_order.append(slot.player.id)
if slot.position:
positions[slot.position] = slot.player.id
if not batting_order:
self.notify("Lineup is empty", severity="warning")
return
try:
async with get_session() as session:
await save_lineup(
session,
name=name,
batting_order=batting_order,
positions=positions,
lineup_type="gameday",
)
await session.commit()
self.current_lineup_name = name
await self._load_saved_lineups()
self.notify(f"Saved: {name}")
except Exception as e:
logger.error(f"Failed to save lineup: {e}")
self.notify(f"Error saving: {e}", severity="error")
async def _do_load_lineup(self, name: str) -> None:
"""Load a lineup by name."""
try:
async with get_session() as session:
lineup = await get_lineup_by_name(session, name)
if not lineup:
self.notify(f"Lineup '{name}' not found", severity="error")
return
self._init_lineup_slots()
player_lookup = {p.id: p for p in self.my_batters}
batting_order = lineup.batting_order or []
positions_map = lineup.positions or {}
player_positions = {v: k for k, v in positions_map.items()}
for i, player_id in enumerate(batting_order[:9]):
player = player_lookup.get(player_id)
if player and i < len(self.lineup_slots):
self.lineup_slots[i].player = player
self.lineup_slots[i].position = player_positions.get(player_id)
self.current_lineup_name = name
self.query_one("#lineup-name-input", Input).value = name
self._populate_matchup_table()
self._populate_lineup_table()
self.notify(f"Loaded: {name}")
except Exception as e:
logger.error(f"Failed to load lineup: {e}")
self.notify(f"Error loading: {e}", severity="error")
async def action_clear_lineup(self) -> None:
"""Clear the lineup."""
self._init_lineup_slots()
self._populate_matchup_table()
self._populate_lineup_table()
self.query_one("#lineup-name-input", Input).value = ""
self.current_lineup_name = ""
self.notify("Lineup cleared")

640
src/sba_scout/screens/lineup.py Normal file
View File

@ -0,0 +1,640 @@
"""
Lineup Builder Screen - Set batting order and defensive positions.
Allows managers to:
- Select batters from their roster for the starting lineup
- Set batting order (1-9)
- Assign defensive positions based on eligibility
- Save/load named lineups (vs LHP, vs RHP, etc.)
"""
import logging
from dataclasses import dataclass, field
from typing import ClassVar, Optional
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, ScrollableContainer, Vertical
from textual.screen import Screen
from textual.widgets import (
Button,
DataTable,
Footer,
Header,
Input,
Label,
Select,
Static,
)
from textual.widgets.data_table import RowKey
from ..config import get_settings
from ..db.models import Lineup, Player
from ..db.queries import delete_lineup, get_lineups, get_my_roster, save_lineup, get_lineup_by_name
from ..db.schema import get_session
logger = logging.getLogger(__name__)
# Defensive positions in typical batting order
DEFENSIVE_POSITIONS = ["C", "1B", "2B", "3B", "SS", "LF", "CF", "RF", "DH"]
@dataclass
class LineupSlot:
"""A slot in the batting order with position assignment."""
order: int # 1-9
player: Optional[Player] = None
position: Optional[str] = None
@property
def display_name(self) -> str:
return self.player.name if self.player else "---"
@property
def display_position(self) -> str:
return self.position or "---"
class LineupScreen(Screen):
"""Lineup builder for setting batting order and positions."""
BINDINGS: ClassVar = [
Binding("escape", "app.pop_screen", "Back"),
Binding("q", "app.pop_screen", "Back"),
Binding("a", "add_to_lineup", "Add to Lineup"),
Binding("r", "remove_from_lineup", "Remove"),
Binding("k", "move_up", "Move Up"),
Binding("j", "move_down", "Move Down"),
Binding("p", "cycle_position", "Change Position"),
Binding("s", "save_lineup", "Save"),
Binding("l", "load_lineup", "Load"),
Binding("d", "delete_lineup", "Delete Saved"),
Binding("c", "clear_lineup", "Clear"),
]
# Current lineup state
lineup_slots: list[LineupSlot] = []
available_batters: list[Player] = []
saved_lineups: list[Lineup] = []
current_lineup_name: str = ""
# Track which table has focus for actions
_focus_table: str = "available" # "available" or "lineup"
def compose(self) -> ComposeResult:
"""Compose the lineup builder layout."""
yield Header()
with Horizontal(id="lineup-container"):
# Left panel - Available batters
with Vertical(id="available-panel", classes="lineup-panel"):
yield Label("Available Batters", classes="panel-title")
with ScrollableContainer(id="available-container"):
yield DataTable(
id="available-table",
cursor_type="row",
zebra_stripes=True,
)
yield Static("[a] Add to lineup", classes="hint-text")
# Right panel - Current lineup
with Vertical(id="lineup-panel", classes="lineup-panel"):
yield Label("Current Lineup", classes="panel-title")
with ScrollableContainer(id="lineup-scroll-container"):
yield DataTable(
id="lineup-table",
cursor_type="row",
zebra_stripes=True,
)
yield Static("[k/j] Move Up/Down [p] Change Pos [r] Remove", classes="hint-text")
# Bottom controls
with Horizontal(id="lineup-controls"):
yield Label("Name:")
yield Input(
placeholder="Lineup name (e.g., 'vs LHP')",
id="lineup-name-input",
)
yield Select(
options=[],
prompt="Load saved...",
id="lineup-select",
allow_blank=True,
)
yield Button("Save [s]", id="btn-save", variant="success")
yield Button("Clear [c]", id="btn-clear", variant="warning")
yield Footer()
async def on_mount(self) -> None:
"""Initialize when screen mounts."""
self._init_lineup_slots()
await self._setup_tables()
await self._load_available_batters()
await self._load_saved_lineups()
def _init_lineup_slots(self) -> None:
"""Initialize empty lineup slots for 9 batters."""
self.lineup_slots = [LineupSlot(order=i) for i in range(1, 10)]
async def _setup_tables(self) -> None:
"""Configure both data tables."""
# Available batters table
avail_table = self.query_one("#available-table", DataTable)
avail_table.add_column("Name", key="name", width=20)
avail_table.add_column("Pos", key="pos", width=14)
avail_table.add_column("H", key="hand", width=3)
avail_table.add_column("vL", key="vl", width=5)
avail_table.add_column("vR", key="vr", width=5)
avail_table.add_column("sWAR", key="swar", width=6)
# Lineup table
lineup_table = self.query_one("#lineup-table", DataTable)
lineup_table.add_column("#", key="order", width=3)
lineup_table.add_column("Name", key="name", width=20)
lineup_table.add_column("Pos", key="pos", width=5)
lineup_table.add_column("Elig", key="elig", width=14)
lineup_table.add_column("H", key="hand", width=3)
lineup_table.add_column("sWAR", key="swar", width=6)
async def _load_available_batters(self) -> None:
"""Load batters from roster."""
settings = get_settings()
try:
async with get_session() as session:
roster = await get_my_roster(
session,
settings.team.team_abbrev,
settings.team.current_season,
)
# Only major league batters
self.available_batters = [p for p in roster.get("majors", []) if p.is_batter]
self._populate_available_table()
except Exception as e:
logger.error(f"Failed to load batters: {e}")
self.notify(f"Error loading batters: {e}", severity="error")
async def _load_saved_lineups(self) -> None:
"""Load saved lineups for the dropdown."""
try:
async with get_session() as session:
self.saved_lineups = list(await get_lineups(session))
# Populate select
lineup_select = self.query_one("#lineup-select", Select)
options = [(f"{lu.name} ({lu.lineup_type})", lu.name) for lu in self.saved_lineups]
lineup_select.set_options(options)
except Exception as e:
logger.error(f"Failed to load saved lineups: {e}")
def _populate_available_table(self) -> None:
"""Populate the available batters table."""
table = self.query_one("#available-table", DataTable)
table.clear()
# Get players already in lineup
lineup_player_ids = {slot.player.id for slot in self.lineup_slots if slot.player}
# Sort by name
batters = sorted(self.available_batters, key=lambda p: p.name)
for player in batters:
# Skip if already in lineup
if player.id in lineup_player_ids:
continue
positions = ", ".join(player.positions)
hand = player.hand or "?"
# Get ratings from card
vl = "---"
vr = "---"
if player.batter_card:
vl = (
f"{player.batter_card.rating_vl:.0f}" if player.batter_card.rating_vl else "---"
)
vr = (
f"{player.batter_card.rating_vr:.0f}" if player.batter_card.rating_vr else "---"
)
row = (
player.name,
positions,
hand,
vl,
vr,
f"{player.swar:.2f}" if player.swar else "0.00",
)
table.add_row(*row, key=str(player.id))
def _populate_lineup_table(self) -> None:
"""Populate the current lineup table."""
table = self.query_one("#lineup-table", DataTable)
table.clear()
for slot in self.lineup_slots:
if slot.player:
positions = ", ".join(slot.player.positions)
hand = slot.player.hand or "?"
row = (
str(slot.order),
slot.player.name,
slot.display_position,
positions,
hand,
f"{slot.player.swar:.2f}" if slot.player.swar else "0.00",
)
else:
row = (
str(slot.order),
"---",
"---",
"---",
"-",
"---",
)
table.add_row(*row, key=f"slot-{slot.order}")
def _get_selected_available_player(self) -> Optional[Player]:
"""Get the currently selected player from the available table."""
table = self.query_one("#available-table", DataTable)
# Get the cursor row index
cursor_row = table.cursor_row
if cursor_row is None:
return None
try:
# Get all row keys in order
row_keys = list(table.rows.keys())
if cursor_row >= len(row_keys):
return None
# RowKey objects have a .value attribute containing the actual key
row_key = row_keys[cursor_row]
player_id_str = row_key.value # This is the string we passed to add_row(key=...)
player_id = int(player_id_str)
return next((p for p in self.available_batters if p.id == player_id), None)
except (IndexError, ValueError, KeyError, AttributeError) as e:
logger.error(f"Failed to get selected player: {e}")
return None
def _get_selected_lineup_slot(self) -> Optional[LineupSlot]:
"""Get the currently selected lineup slot."""
table = self.query_one("#lineup-table", DataTable)
if table.cursor_row is None:
return None
if 0 <= table.cursor_row < len(self.lineup_slots):
return self.lineup_slots[table.cursor_row]
return None
def _find_first_empty_slot(self) -> Optional[LineupSlot]:
"""Find the first empty slot in the lineup."""
for slot in self.lineup_slots:
if slot.player is None:
return slot
return None
def _suggest_position(self, player: Player) -> str:
"""Suggest the best available position for a player."""
positions = player.positions
# Positions already filled in lineup
filled_positions = {
slot.position for slot in self.lineup_slots if slot.position and slot.player
}
# Find first eligible position not yet filled
for pos in positions:
if pos in DEFENSIVE_POSITIONS and pos not in filled_positions:
return pos
# Default to DH if nothing else available
if "DH" not in filled_positions:
return "DH"
# Just use first position if all else fails
return positions[0] if positions else "DH"
def _get_next_position(self, player: Player, current: Optional[str]) -> str:
"""Get next valid position for cycling."""
positions = player.positions
# Add DH if not already there
if "DH" not in positions:
positions = list(positions) + ["DH"]
if not current or current not in positions:
return positions[0] if positions else "DH"
# Find next position in cycle
idx = positions.index(current)
return positions[(idx + 1) % len(positions)]
# =========================================================================
# Event Handlers
# =========================================================================
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
"""Track which table has focus."""
if event.data_table.id == "available-table":
self._focus_table = "available"
elif event.data_table.id == "lineup-table":
self._focus_table = "lineup"
async def on_select_changed(self, event: Select.Changed) -> None:
"""Handle lineup selection from dropdown."""
if event.select.id == "lineup-select":
if event.value and event.value != Select.BLANK:
await self._do_load_lineup(event.value)
async def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button clicks."""
if event.button.id == "btn-save":
await self.action_save_lineup()
elif event.button.id == "btn-clear":
await self.action_clear_lineup()
# =========================================================================
# Actions
# =========================================================================
async def action_add_to_lineup(self) -> None:
"""Add selected available batter to lineup."""
player = self._get_selected_available_player()
if not player:
self.notify("Select a player from available batters", severity="warning")
return
slot = self._find_first_empty_slot()
if not slot:
self.notify("Lineup is full (9 batters)", severity="warning")
return
# Add player with suggested position
slot.player = player
slot.position = self._suggest_position(player)
# Remember which slot the player was added to (0-indexed)
added_slot_idx = slot.order - 1
# Refresh both tables
self._populate_available_table()
self._populate_lineup_table()
# Reset available batters cursor to top
avail_table = self.query_one("#available-table", DataTable)
if len(list(avail_table.rows)) > 0:
avail_table.move_cursor(row=0)
# Move lineup cursor to the newly added player and focus that table
lineup_table = self.query_one("#lineup-table", DataTable)
lineup_table.move_cursor(row=added_slot_idx)
lineup_table.focus()
self.notify(f"Added {player.name} at {slot.position}")
async def action_remove_from_lineup(self) -> None:
"""Remove selected player from lineup."""
lineup_table = self.query_one("#lineup-table", DataTable)
cursor_row = lineup_table.cursor_row # Save cursor position
slot = self._get_selected_lineup_slot()
if not slot or not slot.player:
self.notify("Select a player in the lineup to remove", severity="warning")
return
player_name = slot.player.name
slot.player = None
slot.position = None
# Refresh both tables
self._populate_available_table()
self._populate_lineup_table()
# Restore cursor position
if cursor_row is not None:
lineup_table.move_cursor(row=cursor_row)
self.notify(f"Removed {player_name}")
async def action_move_up(self) -> None:
"""Move selected lineup slot up in batting order."""
slot = self._get_selected_lineup_slot()
if not slot:
return
idx = slot.order - 1 # 0-indexed
if idx <= 0:
self.notify("Already at top of order")
return
# Swap with slot above
self.lineup_slots[idx], self.lineup_slots[idx - 1] = (
self.lineup_slots[idx - 1],
self.lineup_slots[idx],
)
# Update order numbers
for i, s in enumerate(self.lineup_slots):
s.order = i + 1
# Refresh and move cursor
self._populate_lineup_table()
table = self.query_one("#lineup-table", DataTable)
table.move_cursor(row=idx - 1)
async def action_move_down(self) -> None:
"""Move selected lineup slot down in batting order."""
slot = self._get_selected_lineup_slot()
if not slot:
return
idx = slot.order - 1 # 0-indexed
if idx >= 8: # Already at position 9
self.notify("Already at bottom of order")
return
# Swap with slot below
self.lineup_slots[idx], self.lineup_slots[idx + 1] = (
self.lineup_slots[idx + 1],
self.lineup_slots[idx],
)
# Update order numbers
for i, s in enumerate(self.lineup_slots):
s.order = i + 1
# Refresh and move cursor
self._populate_lineup_table()
table = self.query_one("#lineup-table", DataTable)
table.move_cursor(row=idx + 1)
async def action_cycle_position(self) -> None:
"""Cycle through available positions for selected player."""
table = self.query_one("#lineup-table", DataTable)
cursor_row = table.cursor_row # Save cursor position
slot = self._get_selected_lineup_slot()
if not slot or not slot.player:
self.notify("Select a player in the lineup", severity="warning")
return
slot.position = self._get_next_position(slot.player, slot.position)
self._populate_lineup_table()
# Restore cursor position
if cursor_row is not None:
table.move_cursor(row=cursor_row)
self.notify(f"{slot.player.name} -> {slot.position}")
async def action_save_lineup(self) -> None:
"""Save current lineup."""
name_input = self.query_one("#lineup-name-input", Input)
name = name_input.value.strip()
if not name:
self.notify("Enter a lineup name", severity="warning")
name_input.focus()
return
# Build batting order and positions
batting_order = []
positions = {}
for slot in self.lineup_slots:
if slot.player:
batting_order.append(slot.player.id)
if slot.position:
positions[slot.position] = slot.player.id
if not batting_order:
self.notify("Lineup is empty", severity="warning")
return
try:
async with get_session() as session:
await save_lineup(
session,
name=name,
batting_order=batting_order,
positions=positions,
lineup_type="standard",
description=f"Saved lineup with {len(batting_order)} batters",
)
await session.commit()
self.current_lineup_name = name
await self._load_saved_lineups() # Refresh dropdown
self.notify(f"Saved lineup: {name}", severity="information")
except Exception as e:
logger.error(f"Failed to save lineup: {e}")
self.notify(f"Error saving lineup: {e}", severity="error")
async def action_load_lineup(self) -> None:
"""Focus the lineup dropdown for selection."""
self.query_one("#lineup-select", Select).focus()
async def _do_load_lineup(self, name: str) -> None:
"""Load a specific lineup by name."""
try:
async with get_session() as session:
lineup = await get_lineup_by_name(session, name)
if not lineup:
self.notify(f"Lineup '{name}' not found", severity="error")
return
# Reset slots
self._init_lineup_slots()
# Build player lookup
player_lookup = {p.id: p for p in self.available_batters}
# Load batting order
batting_order = lineup.batting_order or []
positions_map = lineup.positions or {}
# Reverse map: player_id -> position
player_positions = {v: k for k, v in positions_map.items()}
for i, player_id in enumerate(batting_order[:9]):
player = player_lookup.get(player_id)
if player and i < len(self.lineup_slots):
self.lineup_slots[i].player = player
self.lineup_slots[i].position = player_positions.get(player_id)
# Update UI
self.current_lineup_name = name
name_input = self.query_one("#lineup-name-input", Input)
name_input.value = name
self._populate_available_table()
self._populate_lineup_table()
self.notify(f"Loaded lineup: {name}")
except Exception as e:
logger.error(f"Failed to load lineup: {e}")
self.notify(f"Error loading lineup: {e}", severity="error")
async def action_clear_lineup(self) -> None:
"""Clear all slots in the current lineup."""
self._init_lineup_slots()
self._populate_available_table()
self._populate_lineup_table()
# Clear name input
name_input = self.query_one("#lineup-name-input", Input)
name_input.value = ""
self.current_lineup_name = ""
self.notify("Lineup cleared")
async def action_delete_lineup(self) -> None:
"""Delete the currently loaded lineup from the database."""
name_input = self.query_one("#lineup-name-input", Input)
name = name_input.value.strip()
if not name:
self.notify("Enter or load a lineup name to delete", severity="warning")
return
# Check if this lineup exists
lineup_exists = any(lu.name == name for lu in self.saved_lineups)
if not lineup_exists:
self.notify(f"Lineup '{name}' not found in saved lineups", severity="warning")
return
try:
async with get_session() as session:
deleted = await delete_lineup(session, name)
await session.commit()
if deleted:
# Clear current lineup if it was the one deleted
if self.current_lineup_name == name:
self._init_lineup_slots()
self._populate_available_table()
self._populate_lineup_table()
self.current_lineup_name = ""
name_input.value = ""
await self._load_saved_lineups() # Refresh dropdown
self.notify(f"Deleted lineup: {name}", severity="information")
else:
self.notify(f"Lineup '{name}' not found", severity="error")
except Exception as e:
logger.error(f"Failed to delete lineup: {e}")
self.notify(f"Error deleting lineup: {e}", severity="error")

341
src/sba_scout/screens/matchup.py Normal file
View File

@ -0,0 +1,341 @@
"""
Matchup Scout Screen - Analyze batters vs opposing pitchers.
Core scouting feature that helps set optimal lineups based on
batter/pitcher handedness matchups.
"""
import logging
from typing import ClassVar
from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, ScrollableContainer, Vertical
from textual.screen import Screen
from textual.widgets import (
DataTable,
Footer,
Header,
Label,
Select,
Static,
)
from ..calc.matchup import MatchupResult, calculate_team_matchups_cached
from ..calc.score_cache import ensure_cache_exists
from ..config import get_settings
from ..db.models import Player, Team
from ..db.queries import get_all_teams, get_my_roster, get_pitchers, get_team_by_id
from ..db.schema import get_session
logger = logging.getLogger(__name__)
class MatchupScreen(Screen):
"""Matchup scouting screen for analyzing batters vs pitchers."""
BINDINGS: ClassVar = [
Binding("escape", "app.pop_screen", "Back"),
Binding("q", "app.pop_screen", "Back"),
Binding("r", "refresh_matchups", "Refresh"),
Binding("s", "sort_by_rating", "Sort: Rating", show=True),
Binding("n", "sort_by_name", "Sort: Name", show=True),
Binding("p", "sort_by_position", "Sort: Position", show=True),
]
# Current state
teams: list[Team] = []
opponent_pitchers: list[Player] = []
my_batters: list[Player] = []
selected_team_id: int | None = None
selected_pitcher: Player | None = None
matchup_results: list[MatchupResult] = []
current_sort: str = "rating" # "rating", "name", or "position"
def compose(self) -> ComposeResult:
"""Compose the matchup scout layout."""
yield Header()
with Vertical(id="matchup-container"):
# Selection row
with Horizontal(id="matchup-selectors"):
yield Label("Opponent:", id="opponent-label")
yield Select(
options=[],
prompt="Select Team",
id="team-select",
allow_blank=True,
)
yield Label("Pitcher:", id="pitcher-label")
yield Select(
options=[],
prompt="Select Pitcher",
id="pitcher-select",
allow_blank=True,
)
# Pitcher info header
yield Static("Select an opponent team and pitcher", id="pitcher-info")
# Matchup table
with ScrollableContainer(id="matchup-table-container"):
yield DataTable(id="matchup-table", cursor_type="none", zebra_stripes=True)
yield Footer()
async def on_mount(self) -> None:
"""Load initial data when screen mounts."""
await self._setup_table()
await self._ensure_score_cache()
await self._load_teams()
await self._load_my_batters()
async def _ensure_score_cache(self) -> None:
"""Ensure standardized score cache exists, building if necessary."""
try:
async with get_session() as session:
await ensure_cache_exists(session)
except Exception as e:
logger.error(f"Failed to ensure score cache: {e}")
self.notify(f"Error building score cache: {e}", severity="error")
async def _setup_table(self) -> None:
"""Configure the matchup data table columns."""
table = self.query_one("#matchup-table", DataTable)
# Columns: suggested order, name, hand, positions, rating, tier, sWAR
columns = [
("#", 3, ""), # Suggested batting order position
("Name", 20, ""),
("H", 3, ""), # Batter's hand
("Pos", 14, ""), # Positions
("Rating", 8, ""), # Matchup rating
("Tier", 5, ""), # Letter grade
("sWAR", 6, ""),
("Split", 5, ""), # Which split was used (vL/vR)
]
for name, width, prefix in columns:
label = f"{prefix}{name}"
table.add_column(label, key=name.lower(), width=width)
async def _load_teams(self) -> None:
"""Load all teams for the opponent selector."""
settings = get_settings()
try:
async with get_session() as session:
self.teams = list(
await get_all_teams(session, settings.team.current_season, active_only=True)
)
# Populate team select - exclude my team
team_select = self.query_one("#team-select", Select)
options = [
(f"{t.abbrev} - {t.short_name}", t.id)
for t in self.teams
if t.abbrev != settings.team.team_abbrev
]
team_select.set_options(options)
except Exception as e:
logger.error(f"Failed to load teams: {e}")
self.notify(f"Error loading teams: {e}", severity="error")
async def _load_my_batters(self) -> None:
"""Load my team's batters for matchup analysis."""
settings = get_settings()
try:
async with get_session() as session:
roster = await get_my_roster(
session,
settings.team.team_abbrev,
settings.team.current_season,
)
# Only include major league batters
self.my_batters = [p for p in roster.get("majors", []) if p.is_batter]
except Exception as e:
logger.error(f"Failed to load batters: {e}")
self.notify(f"Error loading batters: {e}", severity="error")
async def _load_opponent_pitchers(self, team_id: int) -> None:
"""Load pitchers for the selected opponent team."""
settings = get_settings()
try:
async with get_session() as session:
self.opponent_pitchers = list(
await get_pitchers(
session, team_id=team_id, season=settings.team.current_season
)
)
# Populate pitcher select
pitcher_select = self.query_one("#pitcher-select", Select)
options = [(f"{p.name} ({p.hand or '?'}HP)", p.id) for p in self.opponent_pitchers]
pitcher_select.set_options(options)
# Clear current pitcher selection
self.selected_pitcher = None
self.query_one("#pitcher-info", Static).update("Select a pitcher to see matchups")
self._clear_table()
except Exception as e:
logger.error(f"Failed to load pitchers: {e}")
self.notify(f"Error loading pitchers: {e}", severity="error")
def _clear_table(self) -> None:
"""Clear the matchup table."""
table = self.query_one("#matchup-table", DataTable)
table.clear()
self.matchup_results = []
async def _calculate_and_display_matchups(self) -> None:
"""Calculate matchups and populate the table."""
if not self.selected_pitcher:
return
pitcher_hand = self.selected_pitcher.hand or "R"
if pitcher_hand not in ("L", "R"):
pitcher_hand = "R" # Default to R if unknown
# Get pitcher's card data
pitcher_card = self.selected_pitcher.pitcher_card
# Calculate matchups using cached scores
async with get_session() as session:
self.matchup_results = await calculate_team_matchups_cached(
session,
self.my_batters,
self.selected_pitcher,
pitcher_card,
)
# Apply current sort
self._apply_sort()
# Update pitcher info - show if pitcher has card data
pitcher_info = self.query_one("#pitcher-info", Static)
positions = ", ".join(self.selected_pitcher.positions)
card_status = "" if pitcher_card else " [no card]"
pitcher_info.update(
f"vs {self.selected_pitcher.name} ({pitcher_hand}HP) - {positions}{card_status}"
)
# Populate table
self._populate_table()
def _apply_sort(self) -> None:
"""Sort matchup results based on current sort mode."""
if self.current_sort == "rating":
# Sort by rating descending (None at end)
def rating_key(r: MatchupResult) -> tuple[int, float]:
if r.rating is None:
return (1, 0)
return (0, -r.rating)
self.matchup_results.sort(key=rating_key)
elif self.current_sort == "name":
self.matchup_results.sort(key=lambda r: r.player.name)
elif self.current_sort == "position":
# Sort by primary position, then name
def pos_key(r: MatchupResult) -> tuple[str, str]:
positions = r.player.positions
primary = positions[0] if positions else "ZZ"
return (primary, r.player.name)
self.matchup_results.sort(key=pos_key)
def _populate_table(self) -> None:
"""Populate the table with matchup results."""
table = self.query_one("#matchup-table", DataTable)
table.clear()
for i, result in enumerate(self.matchup_results, start=1):
player = result.player
positions = ", ".join(player.positions)
row = (
str(i), # Suggested order
player.name,
result.batter_hand,
positions,
result.rating_display,
result.tier,
f"{player.swar:.2f}" if player.swar else "0.00",
result.split_display, # e.g., "vR/vL" showing batter/pitcher splits
)
table.add_row(*row, key=str(player.id))
# =========================================================================
# Event Handlers
# =========================================================================
async def on_select_changed(self, event: Select.Changed) -> None:
"""Handle selection changes in dropdowns."""
if event.select.id == "team-select":
if event.value and event.value != Select.BLANK:
self.selected_team_id = event.value
await self._load_opponent_pitchers(event.value)
else:
self.selected_team_id = None
# Clear pitcher select
pitcher_select = self.query_one("#pitcher-select", Select)
pitcher_select.set_options([])
self._clear_table()
self.query_one("#pitcher-info", Static).update(
"Select an opponent team and pitcher"
)
elif event.select.id == "pitcher-select":
if event.value and event.value != Select.BLANK:
# Find the selected pitcher
self.selected_pitcher = next(
(p for p in self.opponent_pitchers if p.id == event.value),
None,
)
await self._calculate_and_display_matchups()
else:
self.selected_pitcher = None
self._clear_table()
self.query_one("#pitcher-info", Static).update("Select a pitcher to see matchups")
# =========================================================================
# Actions
# =========================================================================
async def action_refresh_matchups(self) -> None:
"""Refresh matchup calculations."""
if self.selected_pitcher:
await self._calculate_and_display_matchups()
self.notify("Matchups refreshed", severity="information")
else:
self.notify("Select a pitcher first", severity="warning")
def action_sort_by_rating(self) -> None:
"""Sort table by matchup rating."""
self.current_sort = "rating"
if self.matchup_results:
self._apply_sort()
self._populate_table()
self.notify("Sorted by rating")
def action_sort_by_name(self) -> None:
"""Sort table by player name."""
self.current_sort = "name"
if self.matchup_results:
self._apply_sort()
self._populate_table()
self.notify("Sorted by name")
def action_sort_by_position(self) -> None:
"""Sort table by position."""
self.current_sort = "position"
if self.matchup_results:
self._apply_sort()
self._populate_table()
self.notify("Sorted by position")

2
src/sba_scout/screens/roster.py
View File

@ -135,7 +135,7 @@ class RosterScreen(Screen):
roster = await get_my_roster( roster = await get_my_roster(
session, session,
settings.team.team_abbrev, settings.team.team_abbrev,
13, # TODO: Get current season from config/API settings.team.current_season,
) )
# Calculate sWAR totals # Calculate sWAR totals