Capstone - Complete Analytics System

# Python: Understanding Event Data Structure
import pandas as pd
import json
from statsbombpy import sb

# Load StatsBomb event data
events = sb.events(match_id=3788741)

# Examine the structure
print(events.info())
print(events.head())

# Core fields in event data
core_fields = [
    "id",           # Unique event identifier
    "index",        # Sequential order in match
    "period",       # 1 = first half, 2 = second half
    "timestamp",    # Time in period (HH:MM:SS.mmm)
    "minute",       # Match minute
    "second",       # Second within minute
    "type",         # Event type (Pass, Shot, etc.)
    "player",       # Player performing action
    "team",         # Team in possession
    "location",     # [x, y] coordinates
    "duration"      # Event duration in seconds
]

# View available columns
print("Available columns:")
print(events.columns.tolist())

# Count events by type
event_counts = events["type"].value_counts()
print("\nEvent counts by type:")
print(event_counts)

# Python: Exploring Event Qualifiers
import pandas as pd

# Extract pass-specific fields
passes = events[events["type"] == "Pass"].copy()

# Common pass columns
pass_columns = [
    "id", "minute", "second", "player", "team",
    "location", "pass_end_location",
    "pass_length", "pass_angle", "pass_height",
    "pass_body_part", "pass_type",
    "pass_outcome", "pass_recipient",
    "pass_cross", "pass_through_ball", "pass_switch"
]

# Filter to available columns
available_pass_cols = [c for c in pass_columns if c in passes.columns]
pass_data = passes[available_pass_cols]

# View pass qualifiers summary
pass_summary = {
    "total_passes": len(passes),
    "crosses": passes["pass_cross"].sum() if "pass_cross" in passes.columns else 0,
    "through_balls": passes["pass_through_ball"].sum() if "pass_through_ball" in passes.columns else 0,
    "successful": passes["pass_outcome"].isna().sum(),
    "unsuccessful": passes["pass_outcome"].notna().sum()
}

print("Pass Summary:")
for k, v in pass_summary.items():
    print(f"  {k}: {v}")

# Extract shot-specific fields
shots = events[events["type"] == "Shot"].copy()

# Examine shot data
print("\nShot columns available:")
shot_cols = [c for c in shots.columns if "shot" in c.lower()]
print(shot_cols)

# View shot details
print("\nShot details:")
print(shots[["minute", "player", "shot_statsbomb_xg", "shot_outcome"]].head())

# Python: Parsing Nested JSON Event Data
import json
import pandas as pd
from typing import Dict, Any, List

def flatten_event(event: Dict[str, Any]) -> Dict[str, Any]:
    """Flatten a single nested event into a flat dictionary."""
    # Base fields
    flat = {
        "id": event.get("id"),
        "index": event.get("index"),
        "period": event.get("period"),
        "minute": event.get("minute"),
        "second": event.get("second"),
        "type": event.get("type", {}).get("name"),
        "type_id": event.get("type", {}).get("id"),
        "player": event.get("player", {}).get("name"),
        "player_id": event.get("player", {}).get("id"),
        "team": event.get("team", {}).get("name"),
        "team_id": event.get("team", {}).get("id"),
    }

    # Location
    location = event.get("location")
    if location:
        flat["location_x"] = location[0]
        flat["location_y"] = location[1]

    # Pass-specific fields
    if "pass" in event:
        pass_data = event["pass"]
        flat["pass_length"] = pass_data.get("length")
        flat["pass_angle"] = pass_data.get("angle")
        end_loc = pass_data.get("end_location", [None, None])
        flat["pass_end_x"] = end_loc[0] if end_loc else None
        flat["pass_end_y"] = end_loc[1] if end_loc else None
        flat["pass_outcome"] = pass_data.get("outcome", {}).get("name", "Complete")
        flat["pass_recipient"] = pass_data.get("recipient", {}).get("name")

    # Shot-specific fields
    if "shot" in event:
        shot_data = event["shot"]
        flat["shot_xg"] = shot_data.get("statsbomb_xg")
        flat["shot_outcome"] = shot_data.get("outcome", {}).get("name")
        end_loc = shot_data.get("end_location", [None, None, None])
        flat["shot_end_x"] = end_loc[0] if end_loc else None
        flat["shot_end_y"] = end_loc[1] if end_loc else None

    return flat

def parse_events_file(filepath: str) -> pd.DataFrame:
    """Parse a JSON events file into a flat DataFrame."""
    with open(filepath, "r") as f:
        raw_events = json.load(f)

    flattened = [flatten_event(event) for event in raw_events]
    return pd.DataFrame(flattened)

# Parse events
events_flat = parse_events_file("match_events.json")

print(events_flat.head())
print(f"Parsed {len(events_flat)} events")

# Python: Provider-Agnostic Event Parser
import pandas as pd
from abc import ABC, abstractmethod

class EventParser(ABC):
    """Abstract base class for event data parsers."""

    @abstractmethod
    def parse(self, data) -> pd.DataFrame:
        pass

    @abstractmethod
    def standardize(self, df: pd.DataFrame) -> pd.DataFrame:
        pass

class StatsBombParser(EventParser):
    """Parser for StatsBomb event data."""

    def parse(self, data) -> pd.DataFrame:
        return pd.json_normalize(data)

    def standardize(self, df: pd.DataFrame) -> pd.DataFrame:
        standardized = pd.DataFrame({
            "event_id": range(len(df)),
            "event_type": df["type.name"] if "type.name" in df.columns else df["type"],
            "player": df.get("player.name", df.get("player")),
            "team": df.get("team.name", df.get("team")),
            "minute": df["minute"],
            "second": df["second"],
            "location_x": df["location"].apply(lambda x: x[0] if isinstance(x, list) else None),
            "location_y": df["location"].apply(lambda x: x[1] if isinstance(x, list) else None),
        })
        return standardized

class WyscoutParser(EventParser):
    """Parser for Wyscout event data."""

    def parse(self, data) -> pd.DataFrame:
        return pd.json_normalize(data)

    def standardize(self, df: pd.DataFrame) -> pd.DataFrame:
        standardized = pd.DataFrame({
            "event_id": range(len(df)),
            "event_type": df["eventName"],
            "player": df["playerId"],
            "team": df["teamId"],
            "minute": (df["eventSec"] // 60).astype(int),
            "second": (df["eventSec"] % 60).astype(int),
            "location_x": df["positions"].apply(
                lambda x: x[0]["x"] if x else None
            ),
            "location_y": df["positions"].apply(
                lambda x: x[0]["y"] if x else None
            ),
        })
        return standardized

def get_parser(provider: str) -> EventParser:
    """Factory function to get appropriate parser."""
    parsers = {
        "statsbomb": StatsBombParser(),
        "wyscout": WyscoutParser(),
    }
    return parsers.get(provider.lower())

# Usage
parser = get_parser("statsbomb")
standardized = parser.standardize(events)
print(standardized.head())

# Python: Creating Derived Events
import pandas as pd
import numpy as np

def create_progressive_passes(events: pd.DataFrame) -> pd.DataFrame:
    """Identify progressive passes (advance ball >10m toward goal)."""
    passes = events[events["type"] == "Pass"].copy()

    # Extract coordinates
    passes["start_x"] = passes["location"].apply(lambda x: x[0] if x else None)
    passes["start_y"] = passes["location"].apply(lambda x: x[1] if x else None)
    passes["end_x"] = passes["pass_end_location"].apply(
        lambda x: x[0] if isinstance(x, list) else None
    )
    passes["end_y"] = passes["pass_end_location"].apply(
        lambda x: x[1] if isinstance(x, list) else None
    )

    # Calculate distances to goal (goal at x=120)
    passes["start_dist_to_goal"] = np.sqrt(
        (120 - passes["start_x"])**2 + (40 - passes["start_y"])**2
    )
    passes["end_dist_to_goal"] = np.sqrt(
        (120 - passes["end_x"])**2 + (40 - passes["end_y"])**2
    )

    # Progressive if advances >10m toward goal
    passes["is_progressive"] = (
        passes["start_dist_to_goal"] - passes["end_dist_to_goal"]
    ) > 10

    # Into final third
    passes["into_final_third"] = (passes["start_x"] < 80) & (passes["end_x"] >= 80)

    # Into penalty area
    passes["into_box"] = (
        (passes["end_x"] >= 102) &
        (passes["end_y"] >= 18) &
        (passes["end_y"] <= 62)
    )

    return passes

def create_ball_progressions(events: pd.DataFrame) -> pd.DataFrame:
    """Create combined ball progression events (passes + carries)."""
    progressions = []

    # Progressive passes
    passes = events[events["type"] == "Pass"].copy()
    passes["start_x"] = passes["location"].apply(lambda x: x[0] if x else 0)
    passes["end_x"] = passes["pass_end_location"].apply(
        lambda x: x[0] if isinstance(x, list) else 0
    )
    passes["progress_dist"] = passes["end_x"] - passes["start_x"]
    prog_passes = passes[passes["progress_dist"] > 10].copy()
    prog_passes["progression_type"] = "pass"
    progressions.append(prog_passes)

    # Progressive carries
    if "Carry" in events["type"].values:
        carries = events[events["type"] == "Carry"].copy()
        carries["start_x"] = carries["location"].apply(lambda x: x[0] if x else 0)
        carries["end_x"] = carries["carry_end_location"].apply(
            lambda x: x[0] if isinstance(x, list) else 0
        )
        carries["progress_dist"] = carries["end_x"] - carries["start_x"]
        prog_carries = carries[carries["progress_dist"] > 10].copy()
        prog_carries["progression_type"] = "carry"
        progressions.append(prog_carries)

    return pd.concat(progressions).sort_values("index")

prog_passes = create_progressive_passes(events)
print(f"Progressive passes: {prog_passes['is_progressive'].sum()} "
      f"({prog_passes['is_progressive'].mean()*100:.1f}%)")

# Python: Create Possession Sequences
import pandas as pd
import numpy as np

def create_possession_sequences(events: pd.DataFrame) -> pd.DataFrame:
    """Create possession sequence summaries."""
    events = events.sort_values("index").copy()

    # Identify possession changes
    events["team_change"] = events["team"] != events["team"].shift(1)
    events["new_possession"] = (
        events["team_change"] |
        events["type"].isin(["Starting XI", "Half Start"]) |
        events["team"].shift(1).isna()
    )
    events["possession_id"] = events["new_possession"].cumsum()

    # Helper to safely get location
    def safe_location_x(loc):
        if isinstance(loc, list) and len(loc) > 0:
            return loc[0]
        return np.nan

    events["location_x"] = events["location"].apply(safe_location_x)

    # Summarize each possession
    possession_summary = events.groupby(["possession_id", "team"]).agg(
        start_minute=("minute", "min"),
        start_second=("second", "min"),
        end_minute=("minute", "max"),
        end_second=("second", "max"),
        n_events=("index", "count"),
        n_passes=("type", lambda x: (x == "Pass").sum()),
        has_shot=("type", lambda x: (x == "Shot").any()),
        has_goal=("shot_outcome", lambda x: (x == "Goal").any() if "Goal" in x.values else False),
        start_x=("location_x", "first"),
        end_x=("location_x", "last"),
    ).reset_index()

    # Calculate duration
    possession_summary["duration_seconds"] = (
        (possession_summary["end_minute"] * 60 + possession_summary["end_second"]) -
        (possession_summary["start_minute"] * 60 + possession_summary["start_second"])
    )

    # Classify sequences
    possession_summary["territory_gained"] = (
        possession_summary["end_x"] - possession_summary["start_x"]
    )

    def classify_sequence(row):
        if row["has_goal"]:
            return "Goal"
        elif row["has_shot"]:
            return "Shot"
        elif row["end_x"] >= 102:
            return "Penalty Area Entry"
        elif row["end_x"] >= 80:
            return "Final Third Entry"
        else:
            return "No Threat"

    possession_summary["sequence_quality"] = possession_summary.apply(
        classify_sequence, axis=1
    )

    return possession_summary

possessions = create_possession_sequences(events)
print(possessions["sequence_quality"].value_counts())

# Python: Building Custom Metrics from Events
import pandas as pd
import numpy as np

def calculate_dangerous_passes_received(events: pd.DataFrame) -> pd.DataFrame:
    """Calculate dangerous passes received by player."""
    passes = events[
        (events["type"] == "Pass") &
        (events["pass_outcome"].isna())  # Successful passes
    ].copy()

    # Extract end locations
    passes["end_x"] = passes["pass_end_location"].apply(
        lambda x: x[0] if isinstance(x, list) else None
    )
    passes["end_y"] = passes["pass_end_location"].apply(
        lambda x: x[1] if isinstance(x, list) else None
    )

    # Dangerous zone: final third, central areas
    passes["is_dangerous"] = (
        (passes["end_x"] >= 80) &
        (passes["end_y"] >= 20) &
        (passes["end_y"] <= 60) &
        (passes["pass_recipient"].notna())
    )

    dangerous_received = passes[passes["is_dangerous"]].groupby(
        "pass_recipient"
    ).size().reset_index(name="dangerous_passes_received")

    return dangerous_received.sort_values(
        "dangerous_passes_received", ascending=False
    )

def calculate_pressure_regains(events: pd.DataFrame) -> pd.DataFrame:
    """Calculate pressure regain rate by player."""
    events = events.sort_values("index").copy()

    # Get pressures with next event info
    pressures = events[events["type"] == "Pressure"].copy()
    pressures["next_event"] = events["type"].shift(-1)
    pressures["next_team"] = events["team"].shift(-1)

    # Pressure success: same team regains ball
    pressures["pressure_success"] = (
        (pressures["next_team"] == pressures["team"]) &
        (pressures["next_event"].isin(["Ball Recovery", "Interception"]))
    )

    player_pressure_stats = pressures.groupby("player").agg(
        pressures=("index", "count"),
        pressure_regains=("pressure_success", "sum")
    ).reset_index()

    player_pressure_stats["regain_rate"] = (
        player_pressure_stats["pressure_regains"] /
        player_pressure_stats["pressures"]
    )

    return player_pressure_stats[
        player_pressure_stats["pressures"] >= 5
    ].sort_values("regain_rate", ascending=False)

def calculate_box_entries(events: pd.DataFrame) -> pd.DataFrame:
    """Calculate box entries by player (passes + carries)."""
    actions = events[events["type"].isin(["Pass", "Carry"])].copy()

    def get_end_location(row):
        if row["type"] == "Pass":
            loc = row.get("pass_end_location")
        else:
            loc = row.get("carry_end_location")
        return loc if isinstance(loc, list) else [None, None]

    actions["start_x"] = actions["location"].apply(
        lambda x: x[0] if isinstance(x, list) else None
    )
    actions["start_y"] = actions["location"].apply(
        lambda x: x[1] if isinstance(x, list) else None
    )
    actions["end_loc"] = actions.apply(get_end_location, axis=1)
    actions["end_x"] = actions["end_loc"].apply(lambda x: x[0])
    actions["end_y"] = actions["end_loc"].apply(lambda x: x[1])

    # Check if enters box
    actions["enters_box"] = (
        ((actions["start_x"] < 102) | (actions["start_y"] < 18) | (actions["start_y"] > 62)) &
        (actions["end_x"] >= 102) &
        (actions["end_y"] >= 18) &
        (actions["end_y"] <= 62)
    )

    box_entries = actions[actions["enters_box"]].groupby(
        ["player", "type"]
    ).size().unstack(fill_value=0).reset_index()

    if "Pass" not in box_entries.columns:
        box_entries["Pass"] = 0
    if "Carry" not in box_entries.columns:
        box_entries["Carry"] = 0

    box_entries["total_box_entries"] = box_entries["Pass"] + box_entries["Carry"]

    return box_entries.sort_values("total_box_entries", ascending=False)

# Calculate all metrics
dangerous_received = calculate_dangerous_passes_received(events)
pressure_regains = calculate_pressure_regains(events)
box_entries = calculate_box_entries(events)

print("Top players by dangerous passes received:")
print(dangerous_received.head())

# Python: Event Data Quality Checks
import pandas as pd
import numpy as np
from typing import Dict, Any

def validate_event_data(events: pd.DataFrame) -> Dict[str, Any]:
    """Comprehensive event data quality validation."""
    quality_report = {}

    # 1. Check for missing critical fields
    quality_report["missing_fields"] = {
        "missing_type": events["type"].isna().sum(),
        "missing_player": events["player"].isna().sum(),
        "missing_team": events["team"].isna().sum(),
        "missing_location": events["location"].apply(
            lambda x: x is None or (isinstance(x, list) and len(x) == 0)
        ).sum(),
        "missing_timestamp": events["timestamp"].isna().sum()
    }

    # 2. Validate location coordinates
    def extract_coords(loc):
        if isinstance(loc, list) and len(loc) >= 2:
            return loc[0], loc[1]
        return None, None

    events["loc_x"], events["loc_y"] = zip(*events["location"].apply(extract_coords))

    valid_locs = events.dropna(subset=["loc_x", "loc_y"])
    quality_report["location_issues"] = {
        "x_out_of_bounds": ((valid_locs["loc_x"] < 0) | (valid_locs["loc_x"] > 120)).sum(),
        "y_out_of_bounds": ((valid_locs["loc_y"] < 0) | (valid_locs["loc_y"] > 80)).sum(),
        "suspicious_origin": ((valid_locs["loc_x"] == 0) & (valid_locs["loc_y"] == 0)).sum()
    }

    # 3. Check event sequence consistency
    events_sorted = events.sort_values("index")
    events_sorted["time_seconds"] = events_sorted["minute"] * 60 + events_sorted["second"]
    events_sorted["time_diff"] = events_sorted["time_seconds"].diff()

    quality_report["sequence_issues"] = {
        "backwards_timestamps": (events_sorted["time_diff"] < -5).sum(),
        "duplicate_indices": len(events) - events["index"].nunique()
    }

    # 4. Event count validation
    event_counts = events["type"].value_counts()
    expected_mins = {"Pass": 400, "Shot": 10, "Foul Committed": 15}

    below_expected = []
    for event_type, min_count in expected_mins.items():
        actual = event_counts.get(event_type, 0)
        if actual < min_count:
            below_expected.append({
                "type": event_type,
                "actual": actual,
                "expected_min": min_count
            })

    quality_report["below_expected_counts"] = below_expected

    # 5. Generate overall quality score
    total_events = len(events)
    issues = (
        sum(quality_report["missing_fields"].values()) +
        sum(quality_report["location_issues"].values()) +
        quality_report["sequence_issues"]["backwards_timestamps"]
    )

    quality_report["quality_score"] = 1 - (issues / total_events)

    return quality_report

quality = validate_event_data(events)
print(f"Data Quality Score: {quality['quality_score']*100:.2f}%")
print("\nMissing fields:")
for field, count in quality["missing_fields"].items():
    print(f"  {field}: {count}")

# Python: Optimizing Event Data Processing
import pandas as pd
import numpy as np
from concurrent.futures import ProcessPoolExecutor
import pyarrow.parquet as pq
import pyarrow as pa

# Strategy 1: Use vectorized operations
def process_events_vectorized(events: pd.DataFrame) -> pd.DataFrame:
    """Use vectorized operations instead of apply."""
    # Vectorized location extraction
    locations = pd.DataFrame(
        events["location"].tolist(),
        columns=["x", "y"]
    )
    events = pd.concat([events, locations], axis=1)

    # Vectorized calculations
    events["in_final_third"] = events["x"] > 80
    events["in_box"] = (events["x"] > 102) & (events["y"].between(18, 62))

    return events

# Strategy 2: Process files in chunks
def process_season_chunked(file_paths: list, chunk_size: int = 10):
    """Process large number of files in memory-efficient chunks."""
    results = []

    for i in range(0, len(file_paths), chunk_size):
        chunk_files = file_paths[i:i + chunk_size]

        chunk_data = pd.concat([
            pd.read_json(f) for f in chunk_files
        ])

        # Process chunk
        chunk_summary = summarize_events(chunk_data)
        results.append(chunk_summary)

        # Clear memory
        del chunk_data

    return pd.concat(results)

# Strategy 3: Use Parquet for efficient storage
def save_events_parquet(events: pd.DataFrame, filepath: str):
    """Save events to Parquet format for efficient storage."""
    events.to_parquet(filepath, compression="snappy")

def load_events_parquet(filepath: str, columns: list = None):
    """Load events from Parquet, optionally selecting columns."""
    return pd.read_parquet(filepath, columns=columns)

# Strategy 4: Parallel processing
def process_match(filepath: str) -> pd.DataFrame:
    """Process a single match file."""
    events = pd.read_json(filepath)
    return summarize_events(events)

def process_matches_parallel(file_paths: list, n_workers: int = 4):
    """Process multiple matches in parallel."""
    with ProcessPoolExecutor(max_workers=n_workers) as executor:
        results = list(executor.map(process_match, file_paths))
    return pd.concat(results)

# Benchmark example
import time

# Measure processing time
start = time.time()
processed = process_events_vectorized(events)
elapsed = time.time() - start
print(f"Vectorized processing: {elapsed:.3f} seconds")