📈 Tracking Data Tutorial¶
📋 Step 1: Setup & Prerequisites¶
In [1]:
import requests
import pandas as pd
import json
import numpy as np
# Setup pitch and plot
from mplsoccer import Pitch, VerticalPitch
import os
username = os.environ.get("SKILLCORNER_USERNAME")
password = os.environ.get("SKILLCORNER_PASSWORD")
from skillcorner.client import SkillcornerClient
client = SkillcornerClient(username=username, password=password)
def time_to_seconds(time_str):
if time_str is None:
return 90 * 60 # 120 minutes = 7200 seconds
h, m, s = map(int, time_str.split(":"))
return h * 3600 + m * 60 + s
📥 Step 2: Load Tracking Data¶
In [2]:
####-----------------------------------------------------------------------------------
# If you're a skillcorner client and you know your match_id you can use the pyton client
###------------------------------------------------------------------------------------
# match_id=1886347
# client.save_match_tracking_data(match_id,
# params={'data_version':3},
# filepath=f'{match_id}_tracking_raw.jsonl')
####-----------------------------------------------------------------------------------
# If you've cloned the repo and are in your local machine
###------------------------------------------------------------------------------------
# match_id=1886347
# raw_data=pd.read_json(f'../../../data/matches/{match_id}/{match_id}_tracking_extrapolated.jsonl', lines=True)
####-----------------------------------------------------------------------------------
# If you're on a separate project/environemnt
###------------------------------------------------------------------------------------
match_id = 1886347
tracking_path = f"../../../data/matches/{match_id}/{match_id}_tracking_extrapolated.jsonl"
with open(tracking_path) as f:
first_line = f.readline()
# Tracking files are stored with Git LFS. Without git-lfs, the local file is a
# pointer rather than JSONL, so load from GitHub instead.
if first_line.startswith("version https://git-lfs.github.com/spec/v1"):
tracking_path = (
"https://media.githubusercontent.com/media/SkillCorner/opendata/master/"
f"data/matches/{match_id}/{match_id}_tracking_extrapolated.jsonl"
)
print("Local tracking file is a Git LFS pointer; loading from GitHub instead.")
raw_data = pd.read_json(tracking_path, lines=True)
# This is common
raw_df = pd.json_normalize(
raw_data.to_dict("records"),
"player_data",
["frame", "timestamp", "period", "possession", "ball_data"],
)
# Extract 'player_id' and 'group from the 'possession' dictionary
raw_df["possession_player_id"] = raw_df["possession"].apply(
lambda x: x.get("player_id")
)
raw_df["possession_group"] = raw_df["possession"].apply(lambda x: x.get("group"))
# (Optional) Expand the ball_data with json_normalize
raw_df[["ball_x", "ball_y", "ball_z", "is_detected_ball"]] = pd.json_normalize(
raw_df.ball_data
)
# (Optional) Drop the original 'possession' column if you no longer need it
raw_df = raw_df.drop(columns=["possession", "ball_data"])
# Add the match_id identifier to your dataframe
raw_df["match_id"] = match_id
tracking_df = raw_df.copy()
tracking_df.head()
--------------------------------------------------------------------------- HTTPError Traceback (most recent call last) Cell In[2], line 21 17 ###------------------------------------------------------------------------------------ 18 match_id = 1886347 19 # Construct the raw GitHub URL 20 tracking_data_github_url = f"https://media.githubusercontent.com/media/SkillCorner/opendata/data/matches/{match_id}/{match_id}_tracking_extrapolated.jsonl" # Data is stored using GitLFS ---> 21 raw_data = pd.read_json(tracking_data_github_url, lines=True) 22 23 24 # This is common File ~/AnalyticsCup2_0/opendata/.venv/lib/python3.12/site-packages/pandas/io/json/_json.py:791, in read_json(path_or_buf, orient, typ, dtype, convert_axes, convert_dates, keep_default_dates, precise_float, date_unit, encoding, encoding_errors, lines, chunksize, compression, nrows, storage_options, dtype_backend, engine) 788 if convert_axes is None and orient != "table": 789 convert_axes = True --> 791 json_reader = JsonReader( 792 path_or_buf, 793 orient=orient, 794 typ=typ, 795 dtype=dtype, 796 convert_axes=convert_axes, 797 convert_dates=convert_dates, 798 keep_default_dates=keep_default_dates, 799 precise_float=precise_float, 800 date_unit=date_unit, 801 encoding=encoding, 802 lines=lines, 803 chunksize=chunksize, 804 compression=compression, 805 nrows=nrows, 806 storage_options=storage_options, 807 encoding_errors=encoding_errors, 808 dtype_backend=dtype_backend, 809 engine=engine, 810 ) 812 if chunksize: 813 return json_reader File ~/AnalyticsCup2_0/opendata/.venv/lib/python3.12/site-packages/pandas/io/json/_json.py:904, in JsonReader.__init__(self, filepath_or_buffer, orient, typ, dtype, convert_axes, convert_dates, keep_default_dates, precise_float, date_unit, encoding, lines, chunksize, compression, nrows, storage_options, encoding_errors, dtype_backend, engine) 902 self.data = filepath_or_buffer 903 elif self.engine == "ujson": --> 904 data = self._get_data_from_filepath(filepath_or_buffer) 905 self.data = self._preprocess_data(data) File ~/AnalyticsCup2_0/opendata/.venv/lib/python3.12/site-packages/pandas/io/json/_json.py:944, in JsonReader._get_data_from_filepath(self, filepath_or_buffer) 937 filepath_or_buffer = stringify_path(filepath_or_buffer) 938 if ( 939 not isinstance(filepath_or_buffer, str) 940 or is_url(filepath_or_buffer) 941 or is_fsspec_url(filepath_or_buffer) 942 or file_exists(filepath_or_buffer) 943 ): --> 944 self.handles = get_handle( 945 filepath_or_buffer, 946 "r", 947 encoding=self.encoding, 948 compression=self.compression, 949 storage_options=self.storage_options, 950 errors=self.encoding_errors, 951 ) 952 filepath_or_buffer = self.handles.handle 953 elif ( 954 isinstance(filepath_or_buffer, str) 955 and filepath_or_buffer.lower().endswith( (...) 958 and not file_exists(filepath_or_buffer) 959 ): File ~/AnalyticsCup2_0/opendata/.venv/lib/python3.12/site-packages/pandas/io/common.py:728, in get_handle(path_or_buf, mode, encoding, compression, memory_map, is_text, errors, storage_options) 725 codecs.lookup_error(errors) 727 # open URLs --> 728 ioargs = _get_filepath_or_buffer( 729 path_or_buf, 730 encoding=encoding, 731 compression=compression, 732 mode=mode, 733 storage_options=storage_options, 734 ) 736 handle = ioargs.filepath_or_buffer 737 handles: list[BaseBuffer] File ~/AnalyticsCup2_0/opendata/.venv/lib/python3.12/site-packages/pandas/io/common.py:384, in _get_filepath_or_buffer(filepath_or_buffer, encoding, compression, mode, storage_options) 382 # assuming storage_options is to be interpreted as headers 383 req_info = urllib.request.Request(filepath_or_buffer, headers=storage_options) --> 384 with urlopen(req_info) as req: 385 content_encoding = req.headers.get("Content-Encoding", None) 386 if content_encoding == "gzip": 387 # Override compression based on Content-Encoding header File ~/AnalyticsCup2_0/opendata/.venv/lib/python3.12/site-packages/pandas/io/common.py:289, in urlopen(*args, **kwargs) 283 """ 284 Lazy-import wrapper for stdlib urlopen, as that imports a big chunk of 285 the stdlib. 286 """ 287 import urllib.request --> 289 return urllib.request.urlopen(*args, **kwargs) File /opt/homebrew/Cellar/python@3.12/3.12.12/Frameworks/Python.framework/Versions/3.12/lib/python3.12/urllib/request.py:215, in urlopen(url, data, timeout, cafile, capath, cadefault, context) 213 else: 214 opener = _opener --> 215 return opener.open(url, data, timeout) File /opt/homebrew/Cellar/python@3.12/3.12.12/Frameworks/Python.framework/Versions/3.12/lib/python3.12/urllib/request.py:521, in OpenerDirector.open(self, fullurl, data, timeout) 519 for processor in self.process_response.get(protocol, []): 520 meth = getattr(processor, meth_name) --> 521 response = meth(req, response) 523 return response File /opt/homebrew/Cellar/python@3.12/3.12.12/Frameworks/Python.framework/Versions/3.12/lib/python3.12/urllib/request.py:630, in HTTPErrorProcessor.http_response(self, request, response) 627 # According to RFC 2616, "2xx" code indicates that the client's 628 # request was successfully received, understood, and accepted. 629 if not (200 <= code < 300): --> 630 response = self.parent.error( 631 'http', request, response, code, msg, hdrs) 633 return response File /opt/homebrew/Cellar/python@3.12/3.12.12/Frameworks/Python.framework/Versions/3.12/lib/python3.12/urllib/request.py:559, in OpenerDirector.error(self, proto, *args) 557 if http_err: 558 args = (dict, 'default', 'http_error_default') + orig_args --> 559 return self._call_chain(*args) File /opt/homebrew/Cellar/python@3.12/3.12.12/Frameworks/Python.framework/Versions/3.12/lib/python3.12/urllib/request.py:492, in OpenerDirector._call_chain(self, chain, kind, meth_name, *args) 490 for handler in handlers: 491 func = getattr(handler, meth_name) --> 492 result = func(*args) 493 if result is not None: 494 return result File /opt/homebrew/Cellar/python@3.12/3.12.12/Frameworks/Python.framework/Versions/3.12/lib/python3.12/urllib/request.py:639, in HTTPDefaultErrorHandler.http_error_default(self, req, fp, code, msg, hdrs) 638 def http_error_default(self, req, fp, code, msg, hdrs): --> 639 raise HTTPError(req.full_url, code, msg, hdrs, fp) HTTPError: HTTP Error 404: Not Found
Congrats ! You have your (maybe) first tracking data frame¶
Now we need to get player information !
📂 Step 3: Load Match Metadata¶
In [ ]:
####-----------------------------------------------------------------------------------
# If you're a skillcorner client and you know your match_id you can use the python client
###------------------------------------------------------------------------------------
# match_id=1886347
# raw_match_data=client.get_match(match_id)
####-----------------------------------------------------------------------------------
# If you've cloned the repo and are in your local machine
###------------------------------------------------------------------------------------
# match_id = 1886347
# file_path = f'../../../data/matches/{match_id}/{match_id}_match.json'
# with open(file_path, 'r') as f:
# raw_match_data = json.load(f)
####-----------------------------------------------------------------------------------
# If you're on a separate project/environment
###------------------------------------------------------------------------------------
match_id = 1886347
file_path = f"../../../data/matches/{match_id}/{match_id}_match.json"
with open(file_path, "r") as f:
raw_match_data = json.load(f)
# The output has nested json elements. We process them
raw_match_df = pd.json_normalize(raw_match_data, max_level=2)
raw_match_df["home_team_side"] = raw_match_df["home_team_side"].astype(str)
players_df = pd.json_normalize(
raw_match_df.to_dict("records"),
record_path="players",
meta=[
"home_team_score",
"away_team_score",
"date_time",
"home_team_side",
"home_team.name",
"home_team.id",
"away_team.name",
"away_team.id",
], # data we keep
)
# Take only players who played and create their total time
players_df = players_df[
~((players_df.start_time.isna()) & (players_df.end_time.isna()))
]
players_df["total_time"] = players_df["end_time"].apply(time_to_seconds) - players_df[
"start_time"
].apply(time_to_seconds)
# Create a flag for GK
players_df["is_gk"] = players_df["player_role.acronym"] == "GK"
# Add a flag if the given player is home or away
players_df["match_name"] = (
players_df["home_team.name"] + " vs " + players_df["away_team.name"]
)
# Add a flag if the given player is home or away
players_df["home_away_player"] = np.where(
players_df.team_id == players_df["home_team.id"], "Home", "Away"
)
# Create flag from player
players_df["team_name"] = np.where(
players_df.team_id == players_df["home_team.id"],
players_df["home_team.name"],
players_df["away_team.name"],
)
# Figure out sides
players_df[["home_team_side_1st_half", "home_team_side_2nd_half"]] = (
players_df["home_team_side"]
.astype(str)
.str.strip("[]")
.str.replace("'", "")
.str.split(", ", expand=True)
)
# Clean up sides
players_df["direction_player_1st_half"] = np.where(
players_df.home_away_player == "Home",
players_df.home_team_side_1st_half,
players_df.home_team_side_2nd_half,
)
players_df["direction_player_2nd_half"] = np.where(
players_df.home_away_player == "Home",
players_df.home_team_side_2nd_half,
players_df.home_team_side_1st_half,
)
# Clean up and keep the columns that we want to keep about
columns_to_keep = [
"start_time",
"end_time",
"match_name",
"date_time",
"home_team.name",
"away_team.name",
"id",
"short_name",
"number",
"team_id",
"team_name",
"player_role.position_group",
"total_time",
"player_role.name",
"player_role.acronym",
"is_gk",
"direction_player_1st_half",
"direction_player_2nd_half",
]
players_df = players_df[columns_to_keep]
players_df.head()
We successfully prepared all the meta data we need for our tracking data¶
🔗 Step 4: Merge Dataframes¶
In [ ]:
# Merging datasets
enriched_tracking_data = tracking_df.merge(
players_df, left_on=["player_id"], right_on=["id"]
)
enriched_tracking_data.head()
Now our data is completed¶
🎨 Step 5: Visualize average positions¶
In [ ]:
# Filtering for frames with a team in possession
filtered_df = enriched_tracking_data[
enriched_tracking_data["possession_group"].notnull()
].copy()
# We basically want to convert the X and Y to make sure we're always visualizing left to right. A player's XY will depend on the half so the straight average doesn't work
filtered_df["direction_player"] = np.where(
filtered_df["period"] == 1,
filtered_df["direction_player_1st_half"],
filtered_df["direction_player_2nd_half"],
)
filtered_df["x"] = np.where(
filtered_df["direction_player"] == "right_to_left",
-filtered_df["x"],
filtered_df["x"],
) # Convert X
filtered_df["y"] = np.where(
filtered_df["direction_player"] == "right_to_left",
-filtered_df["y"],
filtered_df["y"],
) # Convert Y
# Create some flags in case we need them later
filtered_df["possession_team_name"] = np.where(
filtered_df["possession_group"] == "home team",
filtered_df["home_team.name"],
filtered_df["away_team.name"],
)
filtered_df["possession_flag"] = np.where(
filtered_df["possession_team_name"] == filtered_df["team_name"], "IP", "OOP"
)
# At this point all of our players have their X and Y adjusted from left to right , so we can aggregate for across the game
aggregated_df = (
filtered_df.groupby(
[
"player_id",
"possession_group",
"team_name",
"possession_team_name",
"possession_flag", # This will allow us to check the position of the player when the team is in or out of possession
"start_time", # With this we can filter in or out the players that came on as subs
"number",
"is_gk",
]
)[["x", "y"]]
.mean()
.reset_index()
)
aggregated_df.sort_values(["possession_flag", "team_name"])
# We have an aggregate by possession type !
In [ ]:
# Visualizing the data on a pitch
pitch = Pitch(
pitch_type="skillcorner",
line_alpha=0.75,
pitch_length=105,
pitch_width=68,
pitch_color="#001400",
line_color="white",
linewidth=1.5,
)
fig, ax = pitch.grid(figheight=8, endnote_height=0, title_height=0)
# What to visualize
possession = "IP" # or 'OOP'
team = aggregated_df.team_name.unique().tolist()[
1
] # Pick one team, you can use the name directly
viz_ip = aggregated_df[
(aggregated_df["possession_flag"] == possession)
& (aggregated_df["team_name"] == team)
].reset_index(drop=True)
ax.scatter(
viz_ip["x"],
viz_ip["y"],
c="#32FE6B",
alpha=0.95,
s=600,
edgecolors="white",
linewidths=2.5,
zorder=10,
label="team",
)
# Annotate player numbers
for i, row in viz_ip.iterrows():
ax.text(
row["x"],
row["y"],
str(row["number"]),
color="black",
fontweight="bold",
fontsize=10,
ha="center",
va="center",
zorder=16,
)
ax.set_title(f"{team} Average Positions in Possession")
fig.savefig("my_visualised_tracking_data.png", dpi=300, format="png")
In [ ]:
# Starters Onlyu
pitch = Pitch(
pitch_type="skillcorner",
line_alpha=0.75,
pitch_length=105,
pitch_width=68,
pitch_color="#001400",
line_color="white",
linewidth=1.5,
)
fig, ax = pitch.grid(figheight=8, endnote_height=0, title_height=0)
# What to visualize
possession = "IP" # or 'OOP'
team = aggregated_df.team_name.unique().tolist()[
1
] # Pick one team, you can use the name directly
viz_ip = aggregated_df[
(aggregated_df["possession_flag"] == possession)
& (aggregated_df["team_name"] == team)
& (aggregated_df["start_time"] == "00:00:00")
].reset_index(drop=True)
ax.scatter(
viz_ip["x"],
viz_ip["y"],
c="#32FE6B",
alpha=0.95,
s=600,
edgecolors="white",
linewidths=2.5,
zorder=10,
label="team",
)
# Annotate player numbers
for i, row in viz_ip.iterrows():
ax.text(
row["x"],
row["y"],
str(row["number"]),
color="black",
fontweight="bold",
fontsize=10,
ha="center",
va="center",
zorder=16,
)
ax.set_title(f"{team} Average Positions in Possession")
fig.savefig("my_visualised_tracking_data_starters.png", dpi=300, format="png")