import pandas as pd import numpy as np from typing import List, Dict, Optional, Tuple from collections import Counter class DataProcessor: def __init__(self): self.players_df = None self.position_limits = { "PG": (1, 4), "SG": (1, 4), "SF": (1, 4), "PF": (1, 4), "C": (1, 4) } def load_data(self, df: pd.DataFrame) -> None: """Load and validate player data""" required_cols = ["Player", "Position", "Credits", "Projected Points", "Team Name", "Lineup Status"] if not all(col in df.columns for col in required_cols): raise ValueError(f"Missing required columns: {required_cols}") self.players_df = df.copy() self._preprocess_data() def _preprocess_data(self) -> None: """Clean and preprocess the data""" # Convert numeric columns numeric_cols = ["Credits", "Projected Points"] for col in numeric_cols: self.players_df[col] = pd.to_numeric(self.players_df[col], errors="coerce").fillna(0) # Add standard deviation if not present if "Std Dev" not in self.players_df.columns: self.players_df["Std Dev"] = self._calculate_std_dev() # Filter out inactive players self.players_df = self.players_df[self.players_df["Lineup Status"] != "Out"] def _calculate_std_dev(self) -> pd.Series: """Calculate standard deviation based on position and projected points""" pos_factors = { "PG": 0.25, "SG": 0.22, "SF": 0.20, "PF": 0.18, "C": 0.15 } return self.players_df.apply( lambda x: x["Projected Points"] * pos_factors.get(x["Position"], 0.20), axis=1 ) def get_boom_bust_analysis(self) -> pd.DataFrame: """Generate boom/bust analysis for all players""" boom_bust_data = [] for _, player in self.players_df.iterrows(): proj_points = player["Projected Points"] std_dev = player["Std Dev"] boom_points = proj_points + 1.5 * std_dev bust_points = max(0, proj_points - 1.5 * std_dev) # Calculate probabilities using normal distribution boom_prob = 1 - 0.9332 # Probability above 1.5 std dev bust_prob = 0.0668 # Probability below -1.5 std dev boom_bust_data.append({ "Player": player["Player"], "Team": player["Team Name"], "Projected Points": proj_points, "Boom Points": round(boom_points, 1), "Bust Points": round(bust_points, 1), "Boom Probability %": round(boom_prob * 100, 1), "Bust Probability %": round(bust_prob * 100, 1) }) return pd.DataFrame(boom_bust_data) def calculate_ownership(self, lineups: List[Dict]) -> pd.DataFrame: """Calculate ownership percentages from generated lineups""" player_count = Counter() total_lineups = len(lineups) # Count player appearances for lineup in lineups: for player in lineup["players"]: player_count[player["Player"]] += 1 ownership_data = [] for _, player in self.players_df.iterrows(): count = player_count.get(player["Player"], 0) ownership_pct = (count / total_lineups * 100) if total_lineups > 0 else 0 value_score = (player["Projected Points"] * 500) / player["Credits"] \ if player["Credits"] > 0 else 0 ownership_data.append({ "Player": player["Player"], "Team": player["Team Name"], "Credits": player["Credits"], "Projected Points": player["Projected Points"], "Count": count, "Ownership %": round(ownership_pct, 2), "Value": round(value_score, 2) }) return pd.DataFrame(ownership_data) def validate_lineup(self, lineup: List[Dict]) -> Tuple[bool, str]: """Validate a lineup against all constraints""" if len(lineup) != 8: return False, "Lineup must have exactly 8 players" # Check salary cap total_salary = sum(player["Salary"] for player in lineup) if not (48500 <= total_salary <= 50000): return False, f"Invalid salary: {total_salary}" # Check position limits pos_counts = Counter(player["Position"] for player in lineup) for pos, (min_count, max_count) in self.position_limits.items(): count = pos_counts.get(pos, 0) if not (min_count <= count <= max_count): return False, f"Invalid {pos} count: {count}" # Check team limits team_counts = Counter(player["Team"] for player in lineup) if max(team_counts.values()) > 5: return False, "Too many players from same team" return True, "Valid lineup"