""" Scouting Score System ===================== Scores South American players combining: - Performance score (60%): weighted metrics per position from PESOS xlsx - Profile score (40%): age, EU nationality, league tier, market value Usage: python scouting_score.py [--min-minutes 450] [--output scouting_scores.csv] """ import re import sys import argparse import pandas as pd import numpy as np import openpyxl from pathlib import Path BASE_DIR = Path(__file__).parent # Default: read the player-season grain from the new pipeline. # --legacy-csv flag still allows reading the old per-row CSV for comparison. PARQUET_PATH = BASE_DIR / 'data' / 'clean' / 'player_season_enriched.parquet' CSV_PATH = BASE_DIR / 'player_stats_p90_all_leagues.csv' WEIGHTS_PATH = BASE_DIR / 'PESOS POR VARIABLE (BLOQUES).xlsx' OUTPUT_PATH = BASE_DIR / 'scouting_scores.csv' # ========================================== # MAPPINGS & CONFIG # ========================================== POSITION_MAP = { 'GK': 'P1', 'DL': 'P2', 'DML': 'P2', 'Wing Back Left': 'P2', 'DR': 'P3', 'DMR': 'P3', 'Wing Back Right': 'P3', 'DC': 'P4', 'Central Defender Centre': 'P4', 'AML': 'P7', 'FWL': 'P7', 'ML': 'P7', 'Winger Left': 'P7', 'DMC': 'P8', 'MC': 'P8', 'Defensive Midfielder Centre': 'P8', 'Defensive Midfielder Left': 'P8', 'Defensive Midfielder Right': 'P8', 'Central Midfielder Left': 'P8', 'Central Midfielder Right': 'P8', 'FW': 'P9', 'Striker Centre/Right': 'P9', 'Striker Left/Centre': 'P9', 'Second Striker Centre': 'P9', 'AMC': 'P10', 'AMR': 'P11', 'FWR': 'P11', 'MR': 'P11', 'Winger Right': 'P11', } METRIC_FIXES = { 'acciones_defensivas_en_ofensiva_rival': 'acciones_defensivas_en_ofensiva', 'presion': 'presiones', } LEAGUE_TIERS = { # Top-5 Europe 'Spain La Liga': 1.0, 'Spanish La Liga': 1.0, 'England Premier League': 1.0, 'Germany Bundesliga': 1.0, 'Italy Serie A': 1.0, 'France Ligue 1': 1.0, # Top continentals (already filter strong rosters) 'Europe Champions League': 1.0, 'Europa League': 0.9, # Other top-flight Europe 'Liga Portugal': 0.8, 'Eredivisie': 0.8, 'Turkey Super Lig': 0.7, 'Jupiler Pro League': 0.7, 'Scotland Premiership': 0.6, 'Russia Premier League': 0.6, 'Austrian Bundesliga': 0.6, 'Croatia Prva HNL': 0.55, 'Danish Superligaen': 0.55, 'Norwegian Eliteserien': 0.55, 'Swedish Allsvenskan': 0.55, 'Polish Ekstraklasa': 0.55, 'Brack Super League': 0.55, 'Serbian Super Liga': 0.5, # 2nd tier Europe 'Bundesliga 2': 0.6, 'England Championship': 0.6, 'Spanish Segunda Division': 0.6, 'Italian Serie B': 0.55, 'French Ligue 2': 0.55, 'Dutch Eerste Divisie': 0.5, 'Belgian Challenger Pro League': 0.45, # 3rd-4th tier Europe 'League One': 0.4, 'League Two England': 0.3, # Cup-style 'Spanish Copa Del Rey': 0.7, # South America (kept from prior) 'Liga Profesional Argentina': 1.0, 'Brasileirao': 1.0, 'Colombia Primera A Apertura': 1.0, 'Ecuador Liga Pro': 1.0, 'Chile Primera': 0.5, 'Colombia Superliga': 0.5, # Youth/intl tournaments 'CONMEBOL Libertadores U20': 0.75, 'CONMEBOL U20': 0.75, 'CONMEBOL U17': 0.75, 'UEFA Under 17 Championship': 0.75, 'UEFA Under 19 Championship': 0.75, 'UEFA Under 21 Championship': 0.8, 'Fifa World Cup': 1.0, # MLS 'MLS': 0.6, # Multi-comp aggregate placeholder (used when n_competitions>1) 'MULTI': 0.7, } DEFAULT_LEAGUE_TIER = 0.5 YOUTH_COMPS = {'CONMEBOL Libertadores U20', 'CONMEBOL U20', 'CONMEBOL U17', 'UEFA Under 17 Championship', 'UEFA Under 19 Championship', 'UEFA Under 21 Championship'} EU_COUNTRIES = { 'Albania', 'Andorra', 'Armenia', 'Austria', 'Azerbaijan', 'Belarus', 'Belgium', 'Bosnia-Herzegovina', 'Bulgaria', 'Croatia', 'Cyprus', 'Czech Republic', 'Denmark', 'England', 'Estonia', 'Finland', 'France', 'Georgia', 'Germany', 'Greece', 'Hungary', 'Iceland', 'Ireland', 'Italy', 'Kazakhstan', 'Kosovo', 'Latvia', 'Liechtenstein', 'Lithuania', 'Luxembourg', 'Malta', 'Moldova', 'Monaco', 'Montenegro', 'Netherlands', 'North Macedonia', 'Norway', 'Poland', 'Portugal', 'Romania', 'Russia', 'San Marino', 'Scotland', 'Serbia', 'Slovakia', 'Slovenia', 'Spain', 'Sweden', 'Switzerland', 'Turkey', 'Ukraine', 'Wales', 'Northern Ireland', 'Türkiye', 'Republic of Ireland', 'Bosnia and Herzegovina', 'Faroe Islands', 'Gibraltar', } PROFILE_WEIGHTS = { 'age': 0.35, 'eu_nationality': 0.25, 'league_tier': 0.20, 'market_value': 0.20, } PERFORMANCE_WEIGHT = 0.60 PROFILE_WEIGHT = 0.40 MISSING_TM_PENALTY = 0.5 # ========================================== # LOAD WEIGHTS # ========================================== def load_weights(path): """Load weights xlsx into {position: {block: [(metric, weight, invertida)]}}""" wb = openpyxl.load_workbook(path) ws = wb['Hoja1'] weights = {} for row in ws.iter_rows(min_row=2, max_row=ws.max_row, values_only=True): pos, block, metric, weight, invertida = row if not pos or not metric: continue metric = METRIC_FIXES.get(metric, metric) inv = str(invertida).strip().lower() == 'true' weights.setdefault(pos, {}).setdefault(block, []).append((metric, weight, inv)) return weights # ========================================== # PERFORMANCE SCORING # ========================================== def compute_percentile_ranks(df, weights): """Percentile ranks per (mapped_position, season) for all relevant metrics.""" metrics_by_pos = {} for pos, blocks in weights.items(): metrics = set() for block_metrics in blocks.values(): for metric, _, _ in block_metrics: metrics.add(metric) metrics_by_pos[pos] = metrics for pos, metrics in metrics_by_pos.items(): mask = df['mapped_position'] == pos if mask.sum() == 0: continue sub = df.loc[mask] for metric in metrics: if metric not in df.columns: continue col_name = f'_rank_{pos}_{metric}' # rank within (position, season): groupby Temporada then rank df.loc[mask, col_name] = ( sub.groupby('Temporada')[metric] .rank(pct=True, na_option='keep') ) return df def compute_performance_scores(df, weights): """Compute performance score for each row — fully vectorized per position group.""" all_block_names = set() for blocks in weights.values(): all_block_names.update(blocks.keys()) # Init block columns for block_name in all_block_names: df[f'block_{block_name}'] = pd.NA df['performance_score'] = pd.NA for pos, blocks in weights.items(): mask = df['mapped_position'] == pos if mask.sum() == 0: continue block_scores_list = [] for block_name, metrics in blocks.items(): col = f'block_{block_name}' # Build weighted sum vectorized numerator = pd.Series(0.0, index=df.index[mask]) denominator = pd.Series(0.0, index=df.index[mask]) for metric, weight, invertida in metrics: rank_col = f'_rank_{pos}_{metric}' if rank_col not in df.columns: continue vals = df.loc[mask, rank_col].astype(float) if invertida: vals = 1.0 - vals valid = vals.notna() numerator += (vals * weight).fillna(0) * valid.astype(float) denominator += weight * valid.astype(float) block_score = numerator / denominator.replace(0, pd.NA) df.loc[mask, col] = block_score block_scores_list.append(block_score) # Performance = mean of block scores if block_scores_list: stacked = pd.concat(block_scores_list, axis=1) df.loc[mask, 'performance_score'] = stacked.mean(axis=1, skipna=True) return df # ========================================== # PROFILE SCORING # ========================================== def parse_market_value(mv_str): """Parse market value string like '€2.80m', '€500k' to float in millions.""" if not mv_str or pd.isna(mv_str) or mv_str == '-': return None mv_str = str(mv_str).strip() match = re.search(r'[€$£]?([\d.,]+)\s*([mkMK]?)', mv_str) if not match: return None num = float(match.group(1).replace(',', '.')) unit = match.group(2).lower() if unit == 'm': return num elif unit == 'k': return num / 1000 return num def has_eu_nationality(nationality_str): """Check if player has any EU nationality.""" if not nationality_str or pd.isna(nationality_str): return False for country in EU_COUNTRIES: if country.lower() in str(nationality_str).lower(): return True return False def compute_profile_scores(df): """Compute profile score components and combined profile score.""" # Age factor def age_factor(age): if pd.isna(age) or age == '': return None try: a = int(float(age)) except (ValueError, TypeError): return None if a < 23: return 1.0 elif a <= 25: return 0.5 return 0.0 # prefer master.age, fall back to tm_age age_src = df.get('age') if age_src is None: age_src = df.get('tm_age', pd.Series(index=df.index, dtype=object)) else: age_src = age_src.where(age_src.notna(), df.get('tm_age')) df['age_factor'] = age_src.apply(age_factor) df['_age_used'] = age_src # EU nationality — prefer master.nationality, fall back to tm_nationality nat_src = df.get('nationality') if nat_src is None: nat_src = df.get('tm_nationality', pd.Series(index=df.index, dtype=object)) else: nat_src = nat_src.where(nat_src.notna() & (nat_src != ''), df.get('tm_nationality')) df['eu_nationality_factor'] = nat_src.apply( lambda x: 1.0 if has_eu_nationality(x) else 0.0 ) df['_nationality_used'] = nat_src # League tier df['league_tier_factor'] = df['Competencia'].map(LEAGUE_TIERS).fillna(DEFAULT_LEAGUE_TIER) # Market value factor def mv_factor(mv_str): mv = parse_market_value(mv_str) if mv is None: return None if mv < 3: return 1.0 elif mv <= 5: return 0.5 return -1 # sentinel for exclusion df['market_value_factor'] = df['tm_market_value'].apply(mv_factor) df['market_value_millions'] = df['tm_market_value'].apply(parse_market_value) # Attribute-data-missing flag — TRUE only if BOTH master and TM attrs absent df['tm_data_missing'] = ( df['_age_used'].isna() | (df['_age_used'].astype(str) == '') | df['_nationality_used'].isna() | (df['_nationality_used'].astype(str) == '') ) # Combined profile score def profile_score(row): if row['tm_data_missing']: numerator = ( PROFILE_WEIGHTS['league_tier'] * row['league_tier_factor'] + PROFILE_WEIGHTS['eu_nationality'] * row['eu_nationality_factor'] ) denominator = PROFILE_WEIGHTS['league_tier'] + PROFILE_WEIGHTS['eu_nationality'] return (numerator / denominator if denominator > 0 else 0) * MISSING_TM_PENALTY components = [] w_total = 0 if row['age_factor'] is not None: components.append(PROFILE_WEIGHTS['age'] * row['age_factor']) w_total += PROFILE_WEIGHTS['age'] components.append(PROFILE_WEIGHTS['eu_nationality'] * row['eu_nationality_factor']) w_total += PROFILE_WEIGHTS['eu_nationality'] components.append(PROFILE_WEIGHTS['league_tier'] * row['league_tier_factor']) w_total += PROFILE_WEIGHTS['league_tier'] if row['market_value_factor'] is not None and row['market_value_factor'] >= 0: components.append(PROFILE_WEIGHTS['market_value'] * row['market_value_factor']) w_total += PROFILE_WEIGHTS['market_value'] return sum(components) / w_total if w_total > 0 else 0 df['profile_score'] = df.apply(profile_score, axis=1) return df # ========================================== # DEDUPLICATION # ========================================== def season_sort_key(s): """Convert season string to sortable int. '24-25' -> 2425, '25' -> 2500.""" s = str(s).strip() if '-' in s: parts = s.split('-') return int(parts[0]) * 100 + int(parts[1]) return int(s) * 100 def deduplicate_players(df): """Keep one row per player: most recent season, prefer domestic, more minutes.""" df['_season_sort'] = df['Temporada'].apply(season_sort_key) df['_is_domestic'] = ~df['Competencia'].isin(YOUTH_COMPS) # Best performance score across all rows best_perf = df.groupby('playerId')['performance_score'].max().rename('best_performance_score') df = df.sort_values( ['_season_sort', '_is_domestic', 'Minutos totales'], ascending=[False, False, False] ) deduped = df.groupby('playerId').first().reset_index() deduped = deduped.merge(best_perf, on='playerId', how='left') deduped.drop(columns=['_season_sort', '_is_domestic'], inplace=True) return deduped # ========================================== # MAIN # ========================================== def main(): parser = argparse.ArgumentParser(description='Scouting Score System') parser.add_argument('--min-minutes', type=int, default=450) parser.add_argument('--output', type=str, default=str(OUTPUT_PATH)) parser.add_argument('--season', type=str, default=None, help='Comma-separated seasons to include, e.g. "25,26"') parser.add_argument('--max-age', type=int, default=None, help='Exclude players older than this (master.age, fallback tm_age)') parser.add_argument('--legacy-csv', action='store_true', help='Read the old per-row CSV instead of the parquet pipeline') args = parser.parse_args() print("Loading data...") if args.legacy_csv: df = pd.read_csv(CSV_PATH) print(f" legacy CSV mode: {len(df)} rows × {len(df.columns)} cols") else: df = pd.read_parquet(PARQUET_PATH) print(f" parquet (player-season grain): {len(df)} rows × {len(df.columns)} cols") weights = load_weights(WEIGHTS_PATH) # Map positions df['mapped_position'] = df['Posicion'].map(POSITION_MAP) unmapped = df[df['mapped_position'].isna() & (df['Posicion'] != '') & (df['Posicion'] != 'Sub')] if len(unmapped) > 0: print(f" Warning: {len(unmapped)} rows with unmapped positions: {unmapped['Posicion'].unique()}") # Filter df = df[df['mapped_position'].notna()].copy() print(f" After position filter: {len(df)} rows") min_mins = df['Competencia'].apply( lambda c: 270 if c in YOUTH_COMPS else args.min_minutes ) df = df[df['Minutos totales'].astype(float) >= min_mins].copy() print(f" After minutes filter (>={args.min_minutes}, >=270 youth): {len(df)} rows") if args.season: valid = {s.strip() for s in args.season.split(',')} df = df[df['Temporada'].astype(str).isin(valid)].copy() print(f" After season filter ({args.season}): {len(df)} rows") # Ensure TM columns exist for col in ['tm_age', 'tm_nationality', 'tm_market_value', 'tm_contract_until']: if col not in df.columns: df[col] = '' # Position group sizes print("\nPosition groups:") for pos in sorted(df['mapped_position'].unique()): n = (df['mapped_position'] == pos).sum() print(f" {pos}: {n} rows") # Performance scoring print("\nComputing percentile ranks...") df = compute_percentile_ranks(df, weights) print("Computing performance scores...") df = compute_performance_scores(df, weights) scored = df['performance_score'].notna().sum() print(f" {scored}/{len(df)} rows scored") # Profile scoring print("\nComputing profile scores...") df = compute_profile_scores(df) # Age filter — uses _age_used (master.age fallback to tm_age) if args.max_age: age_col = pd.to_numeric(df['_age_used'], errors='coerce') over = age_col.notna() & (age_col > args.max_age) if over.sum() > 0: df = df[~over].copy() print(f" Excluded {over.sum()} players over age {args.max_age}") # Market value hard filter (>5M excluded) pre_filter = len(df) df = df[df['market_value_factor'] != -1].copy() excluded = pre_filter - len(df) if excluded > 0: print(f" Excluded {excluded} players with market value >5M") # Final scouting score df['scouting_score'] = ( PERFORMANCE_WEIGHT * df['performance_score'].fillna(0) + PROFILE_WEIGHT * df['profile_score'].fillna(0) ) # Percentile rank within (position, season) df['percentile_rank_in_position'] = ( df.groupby(['mapped_position', 'Temporada'])['performance_score'].rank(pct=True) ) # No dedup — output is at (player, season) grain. # Add best_performance_score as the player's max across seasons in this run. df['best_performance_score'] = df.groupby('playerId')['performance_score'].transform('max') result = df.sort_values('scouting_score', ascending=False).copy() n_unique_players = result['playerId'].nunique() print(f"\n {len(result)} player-seasons across {n_unique_players} unique players") # Build output columns id_cols = ['playerId', 'Jugador', 'Equipo', 'Competencia', 'Temporada', 'n_competitions', 'comps_list', 'Posicion', 'mapped_position', 'Minutos totales', 'Partidos jugados'] attr_cols = [c for c in ['age', '_age_used', 'nationality', '_nationality_used', 'second_nationality', 'foot', 'height', 'weight', 'date_of_birth', 'country_of_birth', 'contract_end_date', 'contract_start_date', 'tm_age', 'tm_nationality', 'tm_market_value', 'tm_contract_until', 'tm_foot', 'tm_height', 'tm_position', 'tm_photo'] if c in result.columns] profile_cols = ['age_factor', 'eu_nationality_factor', 'league_tier_factor', 'market_value_factor', 'market_value_millions', 'profile_score'] block_cols = sorted([c for c in result.columns if c.startswith('block_')]) score_cols = ['performance_score', 'scouting_score', 'percentile_rank_in_position', 'tm_data_missing', 'best_performance_score'] out_cols = id_cols + attr_cols + profile_cols + block_cols + score_cols out_cols = [c for c in out_cols if c in result.columns] output = result[out_cols] output.to_csv(args.output, index=False) print(f"\nOutput written to: {args.output}") print(f"Total players: {len(output)}") # Summary print("\n" + "=" * 80) print("TOP 20 SCOUTING SCORES") print("=" * 80) top = output.head(20) for _, row in top.iterrows(): age = row.get('_age_used', row.get('tm_age', '?')) mv = row.get('tm_market_value', '?') nat = row.get('_nationality_used', row.get('tm_nationality', '?')) season = row.get('Temporada', '?') flag = " [no attrs]" if row.get('tm_data_missing') else "" print(f" {row['scouting_score']:.3f} | {str(row['Jugador'])[:25]:<25} | " f"{str(row['Equipo'])[:20]:<20} | {row['mapped_position']} | s:{season} | " f"age:{age} | {mv} | {nat}{flag}") # Stats print(f"\nScore distribution:") print(f" Mean: {output['scouting_score'].mean():.3f}") print(f" Median: {output['scouting_score'].median():.3f}") print(f" Std: {output['scouting_score'].std():.3f}") print(f" Min: {output['scouting_score'].min():.3f}") print(f" Max: {output['scouting_score'].max():.3f}") tm_missing = output['tm_data_missing'].sum() print(f"\n Players with TM data: {len(output) - tm_missing}") print(f" Players missing TM data: {tm_missing}") if __name__ == '__main__': main()