osama-n097/match-performance-api
0
1"""2pipeline/data_loader.py — Data Loading & Preprocessing3يقابل Notebook 014"""5 6import pandas as pd7import numpy as np8import warnings9from pathlib import Path10from statsbombpy import sb11 12from config import (13 COMPETITION_ID, SEASON_ID, TARGET_TEAM,14 ACTION_TYPE_MAP, DATA_DIR, SEASONS_LIST, SEASON_ID_MAP15)16from utils.uuid_manager import add_uuid_column, add_uuids_to_all17from utils.helpers import ensure_dirs18 19warnings.filterwarnings("ignore")20 21 22# ──────────────────────────────────────────────────────────────────────────────23# 1. LOAD RAW DATA24# ──────────────────────────────────────────────────────────────────────────────25 26def load_matches(competition_id: int = COMPETITION_ID,27 season_id: int = SEASON_ID) -> pd.DataFrame:28 """Load Barcelona matches for a given competition + season"""29 all_matches = sb.matches(30 competition_id=competition_id,31 season_id=season_id32 )33 if all_matches.empty:34 return pd.DataFrame()35 36 barca = all_matches[37 (all_matches["home_team"] == TARGET_TEAM) |38 (all_matches["away_team"] == TARGET_TEAM)39 ].reset_index(drop=True)40 41 barca = add_uuid_column(barca, "uuid", based_on=["match_id"])42 season_label = SEASON_ID_MAP.get(season_id, f"unknown_{season_id}")43 print(f"✅ [{season_label}] Matches loaded: {len(barca)}")44 return barca45 46 47def load_all_events(matches_df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:48 """تحميل Events وLineups لكل الماتشات"""49 all_events, all_lineups = [], []50 51 for idx, row in matches_df.iterrows():52 match_id = row["match_id"]53 events = sb.events(match_id=match_id)54 events["match_id"] = match_id55 all_events.append(events)56 57 lineups = sb.lineups(match_id=match_id)58 for team_name, lineup_df in lineups.items():59 lineup_df["match_id"] = match_id60 lineup_df["team_name"] = team_name61 all_lineups.append(lineup_df)62 63 if (idx + 1) % 5 == 0:64 print(f" Loaded {idx + 1}/{len(matches_df)} matches...")65 66 events_df = pd.concat(all_events, ignore_index=True)67 lineups_df = pd.concat(all_lineups, ignore_index=True)68 69 print(f"✅ Events loaded : {len(events_df):,}")70 print(f"✅ Lineups loaded: {len(lineups_df):,}")71 return events_df, lineups_df72 73 74# ──────────────────────────────────────────────────────────────────────────────75# 2. CLEAN EVENTS76# ──────────────────────────────────────────────────────────────────────────────77 78def _extract_location(loc):79 if isinstance(loc, list) and len(loc) >= 2:80 return loc[0], loc[1]81 return None, None82 83 84def _extract_pass_details(df: pd.DataFrame) -> pd.DataFrame:85 pass_mask = df["type"] == "Pass"86 87 df.loc[pass_mask, "pass_outcome"] = df.loc[pass_mask, "pass_outcome"].apply(88 lambda x: x.get("name", "Complete") if isinstance(x, dict)89 else (x if pd.notna(x) else "Complete")90 )91 df.loc[pass_mask, "pass_end_x"] = df.loc[pass_mask, "pass_end_location"].apply(92 lambda x: x[0] if isinstance(x, list) else None93 )94 df.loc[pass_mask, "pass_end_y"] = df.loc[pass_mask, "pass_end_location"].apply(95 lambda x: x[1] if isinstance(x, list) else None96 )97 df.loc[pass_mask, "bodypart"] = df.loc[pass_mask, "pass_body_part"].apply(98 lambda x: x.get("name", None) if isinstance(x, dict) else x99 )100 101 def is_progressive(row):102 try:103 start_x, end_x = row["location_x"], row["pass_end_x"]104 start_y, end_y = row["location_y"], row["pass_end_y"]105 fwd_dist = end_x - start_x106 # Forward pass threshold: moves ball > 20m toward opposition goal107 passes_threshold = fwd_dist > 20108 # Zone entry: pass ends in final third (x > 80) or penalty area (x > 102, 18 < y < 62)109 enters_final_third = end_x > 80110 enters_penalty_area = end_x > 102 and 18 < end_y < 62111 return int(passes_threshold or enters_penalty_area or (enters_final_third and fwd_dist > 5))112 except:113 return 0114 115 df.loc[pass_mask, "is_progressive_pass"] = df[pass_mask].apply(is_progressive, axis=1)116 df["is_progressive_pass"] = df["is_progressive_pass"].fillna(0).astype(int)117 return df118 119 120def _extract_shot_details(df: pd.DataFrame) -> pd.DataFrame:121 shot_mask = df["type"] == "Shot"122 123 df.loc[shot_mask, "shot_outcome"] = df.loc[shot_mask, "shot_outcome"].apply(124 lambda x: x.get("name", None) if isinstance(x, dict) else x125 )126 df.loc[shot_mask, "shot_xg"] = df.loc[shot_mask, "shot_statsbomb_xg"]127 df.loc[shot_mask, "shot_technique"] = df.loc[shot_mask, "shot_technique"].apply(128 lambda x: x.get("name", None) if isinstance(x, dict) else x129 )130 df.loc[shot_mask, "shot_end_x"] = df.loc[shot_mask, "shot_end_location"].apply(131 lambda x: x[0] if isinstance(x, list) else None132 )133 df.loc[shot_mask, "shot_end_y"] = df.loc[shot_mask, "shot_end_location"].apply(134 lambda x: x[1] if isinstance(x, list) else None135 )136 df.loc[shot_mask, "bodypart"] = df.loc[shot_mask, "shot_body_part"].apply(137 lambda x: x.get("name", None) if isinstance(x, dict) else x138 )139 df.loc[shot_mask, "shot_type_name"] = df.loc[shot_mask, "shot_type"].apply(140 lambda x: x.get("name", None) if isinstance(x, dict) else x141 )142 set_pieces = ["Free Kick", "Corner", "Penalty", "Kick Off"]143 df["shot_after_set_piece"] = df["shot_type_name"].isin(set_pieces).astype(int)144 145 df.loc[shot_mask, "distance_to_goal"] = np.sqrt(146 (120 - df.loc[shot_mask, "location_x"])**2 +147 (40 - df.loc[shot_mask, "location_y"])**2148 )149 df.loc[shot_mask, "angle_to_goal"] = np.abs(150 np.arctan2(df.loc[shot_mask, "location_y"] - 40,151 120 - df.loc[shot_mask, "location_x"])152 )153 return df154 155 156def _extract_carry_details(df: pd.DataFrame) -> pd.DataFrame:157 carry_mask = df["type"] == "Carry"158 df.loc[carry_mask, "carry_end_x"] = df.loc[carry_mask, "carry_end_location"].apply(159 lambda x: x[0] if isinstance(x, list) else None160 )161 df.loc[carry_mask, "carry_end_y"] = df.loc[carry_mask, "carry_end_location"].apply(162 lambda x: x[1] if isinstance(x, list) else None163 )164 return df165 166 167def _extract_dribble_details(df: pd.DataFrame) -> pd.DataFrame:168 dribble_mask = df["type"] == "Dribble"169 df.loc[dribble_mask, "dribble_outcome"] = df.loc[dribble_mask, "dribble_outcome"].apply(170 lambda x: x.get("name", None) if isinstance(x, dict) else x171 )172 return df173 174 175def clean_events(events_df: pd.DataFrame) -> pd.DataFrame:176 """تنظيف وتحضير الـ events"""177 print("🔄 Cleaning events...")178 df = events_df.copy()179 180 # Location181 df["location_x"], df["location_y"] = zip(*df["location"].apply(_extract_location))182 183 # Timestamp184 df["timestamp"] = pd.to_datetime(df["timestamp"], format="%H:%M:%S.%f", errors="coerce")185 df["timestamp_seconds"] = (186 df["timestamp"].dt.hour * 3600 +187 df["timestamp"].dt.minute * 60 +188 df["timestamp"].dt.second189 )190 191 # Flags192 df["under_pressure"] = df["under_pressure"].fillna(False).astype(bool).astype(int)193 df["counterpress"] = df["counterpress"].fillna(False).astype(bool).astype(int)194 195 # Event Index196 df = df.sort_values(["match_id", "index"]).reset_index(drop=True)197 df["event_index"] = df.groupby("match_id").cumcount() + 1198 199 # Details200 df = _extract_pass_details(df)201 df = _extract_shot_details(df)202 df = _extract_carry_details(df)203 df = _extract_dribble_details(df)204 205 # Foul cards206 foul_mask = df["type"] == "Foul Committed"207 if foul_mask.any():208 df.loc[foul_mask, "foul_card"] = df.loc[foul_mask, "foul_committed_card"].apply(209 lambda x: x.get("name", None) if isinstance(x, dict) else x210 )211 212 # Final clean table213 keep_cols = [214 "id", "match_id", "player_id", "player", "team", "team_id",215 "type", "period", "minute", "second", "timestamp_seconds", "event_index",216 "location_x", "location_y", "under_pressure", "counterpress",217 "pass_length", "pass_angle", "pass_outcome", "pass_end_x", "pass_end_y",218 "is_progressive_pass", "bodypart",219 "shot_outcome", "shot_xg", "shot_technique", "shot_end_x", "shot_end_y",220 "shot_after_set_piece", "distance_to_goal", "angle_to_goal",221 "carry_end_x", "carry_end_y", "dribble_outcome",222 "duration",223 ]224 available = [c for c in keep_cols if c in df.columns]225 events_clean = df[available].copy()226 227 # Rename228 events_clean = events_clean.rename(columns={229 "id": "event_id",230 "player": "player_name",231 "team": "team_name",232 "type": "event_type",233 })234 235 # UUID236 if "event_id" in events_clean.columns:237 events_clean = add_uuid_column(events_clean, "uuid", based_on=["event_id"])238 else:239 events_clean = add_uuid_column(events_clean, "uuid")240 241 print(f"✅ Events cleaned: {events_clean.shape}")242 return events_clean243 244 245# ──────────────────────────────────────────────────────────────────────────────246# 3. SPADL CONVERSION247# ──────────────────────────────────────────────────────────────────────────────248 249def build_spadl(events_clean: pd.DataFrame) -> pd.DataFrame:250 """تحويل Events لـ SPADL-like format"""251 print("🔄 Building SPADL actions...")252 253 df = events_clean[254 events_clean["event_type"].isin(ACTION_TYPE_MAP.keys())255 ].copy()256 257 df["type_name"] = df["event_type"].map(ACTION_TYPE_MAP)258 df["result_name"] = df.apply(_get_result, axis=1)259 df["bodypart_name"] = df["bodypart"].fillna("foot")260 df["period_id"] = df["period"]261 df["time_seconds"] = df["timestamp_seconds"]262 df["start_x"] = df["location_x"]263 df["start_y"] = df["location_y"]264 df["end_x"] = df["pass_end_x"].fillna(265 df["carry_end_x"].fillna(266 df["shot_end_x"].fillna(df["location_x"])))267 df["end_y"] = df["pass_end_y"].fillna(268 df["carry_end_y"].fillna(269 df["shot_end_y"].fillna(df["location_y"])))270 271 spadl = df[[272 "match_id", "player_id", "player_name", "team_name",273 "period_id", "time_seconds", "event_index",274 "type_name", "result_name", "bodypart_name",275 "start_x", "start_y", "end_x", "end_y",276 "under_pressure"277 ]].reset_index(drop=True)278 279 spadl = add_uuid_column(spadl, "uuid", based_on=["match_id", "event_index"])280 print(f"✅ SPADL actions: {len(spadl):,}")281 return spadl282 283 284def _get_result(row) -> str:285 etype = row["event_type"]286 if etype == "Pass":287 return "fail" if row.get("pass_outcome") not in [None, "Complete"] else "success"288 if etype == "Shot":289 return "success" if row.get("shot_outcome") == "Goal" else "fail"290 if etype == "Dribble":291 return "success" if row.get("dribble_outcome") == "Complete" else "fail"292 return "success"293 294 295# ──────────────────────────────────────────────────────────────────────────────296# 4. SHOTS FOR xG297# ──────────────────────────────────────────────────────────────────────────────298 299def build_shots_for_xg(events_clean: pd.DataFrame) -> pd.DataFrame:300 """استخراج Shot events جاهزة للـ xG Model"""301 shots = events_clean[events_clean["event_type"] == "Shot"][[302 "event_id", "match_id", "player_id", "player_name",303 "location_x", "location_y", "distance_to_goal", "angle_to_goal",304 "shot_technique", "bodypart", "under_pressure",305 "shot_after_set_piece", "shot_outcome", "shot_xg"306 ]].copy()307 308 shots["is_goal"] = (shots["shot_outcome"] == "Goal").astype(int)309 shots = add_uuid_column(shots, "uuid", based_on=["event_id"])310 print(f"✅ Shots for xG: {len(shots):,}")311 return shots312 313 314# ──────────────────────────────────────────────────────────────────────────────315# 5. SAVE & LOAD316# ──────────────────────────────────────────────────────────────────────────────317 318SEASONS_DIR = DATA_DIR / "seasons"319 320def save_all(matches, events_clean, lineups, spadl, shots_xg):321 ensure_dirs(DATA_DIR)322 matches.to_parquet(DATA_DIR / "matches.parquet", index=False)323 events_clean.to_parquet(DATA_DIR / "events_clean.parquet", index=False)324 lineups.to_parquet(DATA_DIR / "lineups.parquet", index=False)325 spadl.to_parquet(DATA_DIR / "spadl_actions.parquet", index=False)326 shots_xg.to_parquet(DATA_DIR / "shots_for_xg.parquet", index=False)327 print("✅ All data saved to data/")328 329 330def save_season(season_label, matches, events_clean, lineups, spadl, shots_xg):331 """Save per-season data to data/seasons/{season_label}/"""332 season_dir = SEASONS_DIR / season_label.replace("/", "_")333 ensure_dirs(season_dir)334 matches.to_parquet(season_dir / "matches.parquet", index=False)335 events_clean.to_parquet(season_dir / "events_clean.parquet", index=False)336 lineups.to_parquet(season_dir / "lineups.parquet", index=False)337 spadl.to_parquet(season_dir / "spadl_actions.parquet", index=False)338 shots_xg.to_parquet(season_dir / "shots_for_xg.parquet", index=False)339 print(f"✅ [{season_label}] Season data saved to seasons/{season_label.replace('/', '_')}/")340 341 342def load_all() -> dict:343 return {344 "matches": pd.read_parquet(DATA_DIR / "matches.parquet"),345 "events_clean": pd.read_parquet(DATA_DIR / "events_clean.parquet"),346 "lineups": pd.read_parquet(DATA_DIR / "lineups.parquet"),347 "spadl": pd.read_parquet(DATA_DIR / "spadl_actions.parquet"),348 "shots_for_xg": pd.read_parquet(DATA_DIR / "shots_for_xg.parquet"),349 }350 351 352def load_season(season_label: str) -> dict:353 """Load a single season from per-season parquet files."""354 season_dir = SEASONS_DIR / season_label.replace("/", "_")355 return {356 "matches": pd.read_parquet(season_dir / "matches.parquet"),357 "events_clean": pd.read_parquet(season_dir / "events_clean.parquet"),358 "lineups": pd.read_parquet(season_dir / "lineups.parquet"),359 "spadl": pd.read_parquet(season_dir / "spadl_actions.parquet"),360 "shots_for_xg": pd.read_parquet(season_dir / "shots_for_xg.parquet"),361 }362 363 364# ──────────────────────────────────────────────────────────────────────────────365# MAIN366# ──────────────────────────────────────────────────────────────────────────────367 368def run(seasons=None):369 """370 Load data for one or more seasons.371 372 Parameters373 ----------374 seasons : list of (competition_id, season_id, label), optional375 Defaults to all SEASONS_LIST in config.376 """377 if seasons is None:378 seasons = SEASONS_LIST379 380 print("=" * 60)381 print("📊 PIPELINE STEP 1: Data Loading & Preprocessing")382 print(f" Seasons to load: {len(seasons)}")383 print("=" * 60)384 385 all_matches = []386 all_events_clean = []387 all_lineups = []388 all_spadl = []389 all_shots_xg = []390 391 for comp_id, season_id, season_label in seasons:392 print(f"\n── Loading {season_label} (comp={comp_id}, season={season_id}) ──")393 394 matches = load_matches(competition_id=comp_id, season_id=season_id)395 if matches.empty:396 print(f" ⚠️ No Barcelona matches for {season_label}, skipping")397 continue398 399 events_df, lineups_df = load_all_events(matches)400 events_clean = clean_events(events_df)401 lineups_df = add_uuid_column(lineups_df, "uuid",402 based_on=["match_id", "player_id"]403 if "player_id" in lineups_df.columns else None)404 spadl = build_spadl(events_clean)405 shots_xg = build_shots_for_xg(events_clean)406 407 # Add season identifiers408 for df in [matches, events_clean, lineups_df, spadl, shots_xg]:409 df["season_label"] = season_label410 df["season_id"] = season_id411 df["competition_id"] = comp_id412 413 # Save per-season414 save_season(season_label, matches, events_clean, lineups_df, spadl, shots_xg)415 416 all_matches.append(matches)417 all_events_clean.append(events_clean)418 all_lineups.append(lineups_df)419 all_spadl.append(spadl)420 all_shots_xg.append(shots_xg)421 422 # Concatenate all seasons423 if all_matches:424 combined = {425 "matches": pd.concat(all_matches, ignore_index=True),426 "events_clean": pd.concat(all_events_clean, ignore_index=True),427 "lineups": pd.concat(all_lineups, ignore_index=True),428 "spadl": pd.concat(all_spadl, ignore_index=True),429 "shots_xg": pd.concat(all_shots_xg, ignore_index=True),430 }431 save_all(**combined)432 print(f"\n✅ Step 1 Complete!")433 print(f" Seasons loaded: {len(all_matches)}")434 print(f" Matches total : {sum(len(m) for m in all_matches)}")435 print(f" Events total : {sum(len(e) for e in all_events_clean):,}")436 return combined437 438 print("⚠️ No data loaded for any season")439 return None440 441 442if __name__ == "__main__":443 run()444