Spaces:
Sleeping
Sleeping
| 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" |