Capstone - Complete Analytics System

# Python: Define recruitment system architecture
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import date
import pandas as pd
import numpy as np

@dataclass
class Player:
    """Core player entity for recruitment system."""
    id: str
    name: str
    position: str
    age: int
    nationality: str
    current_club: str
    league: str
    contract_expiry: date
    market_value: float  # in millions
    stats: Dict[str, float] = field(default_factory=dict)
    adjusted_stats: Dict[str, float] = field(default_factory=dict)
    notes: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict:
        """Convert player to dictionary."""
        return {
            "id": self.id,
            "name": self.name,
            "position": self.position,
            "age": self.age,
            "nationality": self.nationality,
            "club": self.current_club,
            "league": self.league,
            "contract_expiry": self.contract_expiry,
            "market_value": self.market_value,
            **self.stats
        }

class RecruitmentDatabase:
    """Central database for recruitment system."""

    def __init__(self):
        self.players: Dict[str, Player] = {}
        self.shortlists: Dict[str, List[str]] = {}
        self._dataframe_cache = None

    def add_player(self, player: Player):
        """Add player to database."""
        self.players[player.id] = player
        self._dataframe_cache = None  # Invalidate cache

    def get_player(self, player_id: str) -> Optional[Player]:
        """Retrieve player by ID."""
        return self.players.get(player_id)

    def to_dataframe(self) -> pd.DataFrame:
        """Convert all players to DataFrame."""
        if self._dataframe_cache is None:
            self._dataframe_cache = pd.DataFrame([
                p.to_dict() for p in self.players.values()
            ])
        return self._dataframe_cache

    def create_shortlist(self, name: str):
        """Create new shortlist."""
        self.shortlists[name] = []

    def add_to_shortlist(self, shortlist_name: str, player_id: str):
        """Add player to shortlist."""
        if shortlist_name in self.shortlists:
            self.shortlists[shortlist_name].append(player_id)

print("Player and RecruitmentDatabase classes defined")

# Python: Build data ingestion pipeline
import pandas as pd
import numpy as np
from typing import Optional, Dict
from dataclasses import dataclass
import soccerdata as sd

@dataclass
class DataIngestion:
    """Pipeline for ingesting player data from multiple sources."""

    league_multipliers: Dict[str, float] = None

    def __post_init__(self):
        if self.league_multipliers is None:
            self.league_multipliers = {
                "ENG-Premier League": 1.00,
                "ESP-La Liga": 0.92,
                "ITA-Serie A": 0.90,
                "GER-Bundesliga": 0.90,
                "FRA-Ligue 1": 0.80,
                "NED-Eredivisie": 0.70,
                "POR-Primeira Liga": 0.68,
                "ENG-Championship": 0.65
            }

    def load_fbref_data(self, league: str, season: str = "2023-2024") -> Optional[pd.DataFrame]:
        """Load player data from FBref."""
        try:
            fbref = sd.FBref(leagues=[league], seasons=[season])
            stats = fbref.read_player_season_stats(stat_type="standard")
            return stats.reset_index()
        except Exception as e:
            print(f"Error loading FBref data: {e}")
            return None

    def calculate_per90(self, data: pd.DataFrame,
                        min_minutes: int = 450) -> pd.DataFrame:
        """Normalize statistics to per-90 minutes."""
        df = data[data["Min"] >= min_minutes].copy()

        df["nineties"] = df["Min"] / 90

        per90_cols = ["Gls", "Ast", "xG", "xAG", "npxG"]
        for col in per90_cols:
            if col in df.columns:
                df[f"{col}_p90"] = df[col] / df["nineties"]

        df["goal_contribution_p90"] = (df["Gls"] + df["Ast"]) / df["nineties"]

        return df

    def adjust_for_league(self, data: pd.DataFrame,
                          league: str) -> pd.DataFrame:
        """Apply league quality adjustments."""
        multiplier = self.league_multipliers.get(league, 0.75)
        df = data.copy()

        adj_cols = ["xG_p90", "xAG_p90", "npxG_p90"]
        for col in adj_cols:
            if col in df.columns:
                df[f"adj_{col}"] = df[col] * multiplier

        df["league_quality"] = multiplier
        return df

    def ingest_league(self, league: str,
                      season: str = "2023-2024") -> Optional[pd.DataFrame]:
        """Full ingestion pipeline for a league."""
        data = self.load_fbref_data(league, season)
        if data is None:
            return None

        data = self.calculate_per90(data)
        data = self.adjust_for_league(data, league)
        return data

    def ingest_multiple_leagues(self, leagues: list,
                                season: str = "2023-2024") -> pd.DataFrame:
        """Ingest data from multiple leagues."""
        all_data = []
        for league in leagues:
            data = self.ingest_league(league, season)
            if data is not None:
                data["source_league"] = league
                all_data.append(data)

        return pd.concat(all_data, ignore_index=True) if all_data else pd.DataFrame()

# Example usage
ingestion = DataIngestion()
print("Data ingestion pipeline ready")

# Python: Build position-based scoring system
import pandas as pd
import numpy as np
from typing import Dict, List, Optional
from scipy import stats

class PlayerScorer:
    """Position-based player scoring system."""

    def __init__(self):
        # Position-specific metric weights
        self.position_weights = {
            "Striker": {
                "npxG_p90": 0.30,
                "Gls_p90": 0.25,
                "xAG_p90": 0.10,
                "AerWon_p90": 0.10,
                "Press_p90": 0.10,
                "PrgC_p90": 0.15
            },
            "Winger": {
                "xAG_p90": 0.25,
                "PrgC_p90": 0.20,
                "Succ_p90": 0.15,  # Successful dribbles
                "npxG_p90": 0.15,
                "Crs_p90": 0.15,
                "Press_p90": 0.10
            },
            "Central_Midfielder": {
                "PrgP_p90": 0.25,
                "Cmp%": 0.15,
                "TklW_p90": 0.15,
                "xAG_p90": 0.15,
                "Press_p90": 0.15,
                "npxG_p90": 0.15
            },
            "Center_Back": {
                "AerWon_p90": 0.20,
                "TklW_p90": 0.20,
                "Int_p90": 0.15,
                "PrgP_p90": 0.15,
                "Clr_p90": 0.15,
                "Blocks_p90": 0.15
            }
        }

    def calculate_percentiles(self, data: pd.DataFrame,
                              metrics: List[str]) -> pd.DataFrame:
        """Calculate percentile ranks for metrics."""
        df = data.copy()
        for metric in metrics:
            if metric in df.columns:
                df[f"{metric}_pct"] = stats.rankdata(
                    df[metric], method="average"
                ) / len(df) * 100
        return df

    def score_players(self, data: pd.DataFrame,
                      position: str) -> pd.DataFrame:
        """Score players for a specific position."""
        if position not in self.position_weights:
            raise ValueError(f"Unknown position: {position}")

        weights = self.position_weights[position]
        metrics = list(weights.keys())

        # Calculate percentiles
        df = self.calculate_percentiles(data, metrics)

        # Calculate weighted score
        df["position_score"] = 0
        for metric, weight in weights.items():
            pct_col = f"{metric}_pct"
            if pct_col in df.columns:
                df["position_score"] += weight * df[pct_col]

        # Rank players
        df = df.sort_values("position_score", ascending=False)
        df["rank"] = range(1, len(df) + 1)

        return df

    def generate_position_ranking(self, data: pd.DataFrame,
                                   position: str,
                                   top_n: int = 20) -> pd.DataFrame:
        """Generate top N players for a position."""
        scored = self.score_players(data, position)
        return scored.head(top_n)[[
            "Player", "Squad", "Age", "Min",
            "position_score", "rank"
        ]]

# Example usage
scorer = PlayerScorer()
print("Scoring system initialized with position weights:")
for pos, weights in scorer.position_weights.items():
    print(f"  {pos}: {list(weights.keys())}")

# Python: Build player similarity engine
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Optional, Dict

class SimilarityEngine:
    """Engine for finding similar players."""

    def __init__(self, feature_cols: List[str]):
        self.feature_cols = feature_cols
        self.scaler = StandardScaler()
        self.similarity_matrix = None
        self.player_index = None

    def fit(self, data: pd.DataFrame):
        """Fit the similarity model on player data."""
        # Extract features
        features = data[self.feature_cols].fillna(0)

        # Standardize
        scaled_features = self.scaler.fit_transform(features)

        # Calculate similarity matrix
        self.similarity_matrix = cosine_similarity(scaled_features)

        # Store player index mapping
        self.player_index = {
            player: idx for idx, player in enumerate(data["Player"])
        }

        return self

    def find_similar(self, target_player: str,
                     top_n: int = 10) -> Dict[str, float]:
        """Find most similar players to target."""
        if target_player not in self.player_index:
            raise ValueError(f"Player not found: {target_player}")

        idx = self.player_index[target_player]
        similarities = self.similarity_matrix[idx]

        # Get indices sorted by similarity
        similar_indices = np.argsort(similarities)[::-1]

        # Build result dict (excluding self)
        results = {}
        count = 0
        for i in similar_indices:
            player_name = list(self.player_index.keys())[i]
            if player_name != target_player:
                results[player_name] = similarities[i]
                count += 1
                if count >= top_n:
                    break

        return results

    def find_similar_with_filters(self, target_player: str,
                                   data: pd.DataFrame,
                                   max_age: Optional[int] = None,
                                   max_value: Optional[float] = None,
                                   min_minutes: int = 900,
                                   leagues: Optional[List[str]] = None,
                                   top_n: int = 10) -> pd.DataFrame:
        """Find similar players with filters applied."""
        # Get similarity scores
        all_similar = self.find_similar(target_player, top_n=100)

        # Create results dataframe
        results = pd.DataFrame([
            {"Player": name, "similarity": score}
            for name, score in all_similar.items()
        ])

        # Merge with full data
        results = results.merge(data, on="Player", how="left")

        # Apply filters
        if max_age is not None:
            results = results[results["Age"] <= max_age]
        if max_value is not None:
            results = results[results["market_value"] <= max_value]
        if min_minutes is not None:
            results = results[results["Min"] >= min_minutes]
        if leagues is not None:
            results = results[results["league"].isin(leagues)]

        return results.head(top_n)[[
            "Player", "similarity", "Age", "Squad",
            "market_value", "Min"
        ]]

# Example usage
feature_cols = ["npxG_p90", "xAG_p90", "PrgC_p90", "PrgP_p90", "Press_p90"]
similarity_engine = SimilarityEngine(feature_cols)
print(f"Similarity engine initialized with features: {feature_cols}")

# Python: Replacement analysis workflow
import pandas as pd
from typing import Optional, Tuple

class ReplacementAnalyzer:
    """Analyze and recommend player replacements."""

    def __init__(self, similarity_engine: SimilarityEngine):
        self.similarity_engine = similarity_engine

    def find_replacements(self, departing_player: str,
                          data: pd.DataFrame,
                          budget: float,
                          position_filter: Optional[str] = None,
                          age_range: Tuple[int, int] = (18, 28),
                          top_n: int = 10) -> pd.DataFrame:
        """Find replacement candidates for a departing player."""

        # Get similar players
        candidates = self.similarity_engine.find_similar_with_filters(
            departing_player,
            data,
            max_age=age_range[1],
            max_value=budget,
            top_n=50
        )

        # Filter by position
        if position_filter and "Pos" in candidates.columns:
            candidates = candidates[
                candidates["Pos"].str.contains(position_filter, na=False)
            ]

        # Filter by age range
        candidates = candidates[
            (candidates["Age"] >= age_range[0]) &
            (candidates["Age"] <= age_range[1])
        ]

        # Calculate recommendation scores
        candidates = candidates.copy()
        candidates["value_score"] = (
            candidates["similarity"] * 100 /
            (candidates["market_value"] + 1)
        )
        candidates["development_potential"] = (28 - candidates["Age"]) * 2

        # Normalize scores
        max_value_score = candidates["value_score"].max()
        max_dev = candidates["development_potential"].max()

        candidates["overall_recommendation"] = (
            candidates["similarity"] * 0.5 +
            (candidates["value_score"] / max_value_score) * 0.3 +
            (candidates["development_potential"] / max_dev) * 0.2
        )

        return candidates.sort_values(
            "overall_recommendation", ascending=False
        ).head(top_n)

    def generate_report(self, departing_player: str,
                        replacements: pd.DataFrame) -> str:
        """Generate text report for replacements."""
        report = f"""
=== REPLACEMENT ANALYSIS FOR: {departing_player} ===

TOP 5 RECOMMENDED REPLACEMENTS:
{"-" * 50}
"""
        for i, (_, r) in enumerate(replacements.head(5).iterrows(), 1):
            report += f"""
{i}. {r["Player"]} ({r.get("Squad", "N/A")})
   Age: {r["Age"]} | Value: €{r.get("market_value", 0):.1f}M
   Similarity: {r["similarity"]*100:.1f}% | Value Score: {r["value_score"]:.2f}
   Recommendation Score: {r["overall_recommendation"]:.2f}
"""
        return report

# Example
print("Replacement analysis system ready")

# Python: Generate comprehensive scouting reports
import pandas as pd
import numpy as np
from typing import Dict, List, Optional
from datetime import datetime
import matplotlib.pyplot as plt

class ScoutingReportGenerator:
    """Generate comprehensive scouting reports."""

    def __init__(self, similarity_engine: SimilarityEngine):
        self.similarity_engine = similarity_engine

    def create_radar_chart(self, player_data: Dict,
                           metrics: List[str],
                           title: str) -> plt.Figure:
        """Create radar chart for player profile."""
        # Number of metrics
        N = len(metrics)
        angles = [n / float(N) * 2 * np.pi for n in range(N)]
        angles += angles[:1]  # Complete the loop

        # Get values
        values = [player_data.get(m, 0) for m in metrics]
        values += values[:1]

        # Create plot
        fig, ax = plt.subplots(figsize=(8, 8), subplot_kw=dict(polar=True))
        ax.plot(angles, values, "o-", linewidth=2)
        ax.fill(angles, values, alpha=0.25)
        ax.set_xticks(angles[:-1])
        ax.set_xticklabels(metrics, size=10)
        ax.set_title(title, size=14, fontweight="bold", y=1.08)

        return fig

    def generate_report(self, player_name: str,
                        data: pd.DataFrame) -> str:
        """Generate full scouting report."""

        player = data[data["Player"] == player_name]
        if player.empty:
            raise ValueError(f"Player not found: {player_name}")

        p = player.iloc[0]

        # Build report sections
        header = f"""
================================================================================
                         SCOUTING REPORT
================================================================================
Player: {p.get("Player", "N/A")}
Position: {p.get("Pos", "N/A")} | Age: {p.get("Age", "N/A")}
Current Club: {p.get("Squad", "N/A")} ({p.get("league", "N/A")})
Market Value: €{p.get("market_value", 0):.1f}M
Minutes Played: {p.get("Min", 0)} ({p.get("Min", 0)/90:.1f} 90s)
================================================================================
"""

        performance = f"""
PERFORMANCE METRICS (Per 90 Minutes):
-------------------------------------
Goals: {p.get("Gls_p90", 0):.2f}          | xG: {p.get("xG_p90", 0):.2f}
Assists: {p.get("Ast_p90", 0):.2f}        | xA: {p.get("xAG_p90", 0):.2f}
Non-Penalty xG: {p.get("npxG_p90", 0):.2f}
Goal Contribution: {p.get("goal_contribution_p90", 0):.2f}
"""

        # Get similar players
        try:
            similar = self.similarity_engine.find_similar(player_name, top_n=5)
            similar_text = "\n".join([
                f"  - {name} ({score*100:.1f}% similar)"
                for name, score in similar.items()
            ])
        except:
            similar_text = "  (Unable to calculate)"

        similar_section = f"""
SIMILAR PLAYERS:
----------------
{similar_text}
"""

        recommendation = """
RECOMMENDATION:
---------------
[To be filled by scout based on video analysis]

Strengths:
-

Weaknesses:
-

Fit Assessment:
-

Risk Factors:
-

================================================================================
"""
        report = header + performance + similar_section + recommendation

        # Add metadata
        report += f"""
Report Generated: {datetime.now().strftime("%Y-%m-%d %H:%M")}
Data Source: FBref / StatsBomb
"""

        return report

    def export_report(self, report: str, filename: str):
        """Export report to file."""
        with open(filename, "w") as f:
            f.write(report)
        print(f"Report saved to {filename}")

# Example usage
print("Scouting report generator ready")