サッカーにおけるポジショニング定量化の重要性

サッカーとスポーツアナリティクス、xG（ゴール期待値）によるプレーの定量化

スポーツにおいてデータ分析（スポーツアナリティクス）の重要性は近年増してきており、例えば（野球などに比べ）データ化が難しいと思われてきたサッカー（フットボール）においても、最近はxGといった指標が一般観戦者レベルにまで浸透してきている（xGと比較すると浸透度は低いがVAEP等の他の指標もある）。xGを一言でいえば「ゴール期待値」と訳され、例えば、あるプレーにおいて「今のシュートはxGが0.9であった」と示された場合、そのシュートは非常に得点に期待が持てる（＝決めなければいけない）シチュエーションであったことが示唆される。

これは1試合を通じてのある選手に対しても適用でき、例えば「今日のこの選手のxGは1.3であった」とあれば、その選手は試合を通じて1.3点を決めそうな動きをしていた（＝動きが良かった）ことが分かり、結果1点を獲得していればおおよそ期待通りの成果を出せたと判断できる一方で、もし点数を取れていなかった場合は相手キーパーのファインセーブに遭った、などの理由があるのかもしれないと想像できる。

サッカーは1点が入るか入らないかによる差が大きいので、ある2チームの対戦でxGの指標が2.7対0.3だった、しかし実際には片方のチームがスーパーゴールとファインセーブを重ねたため得点スコアとしては0対1となることがあり得る。この場合、端的に言えば前者のチームはその日は運が悪かったのだと考えられ、必要以上に結果を悲観するべきではないと言える。

ポジショニングの定量化は重要だが進んでいない

一方で、ポジショニング（＝ボールを持っていない（敵味方）選手がどこに立ち位置を取っているか）の重要性は明らかだが、これに対応するスタンダードな概念・スコアは今のところないように見える。

例えば、「攻撃側の選手がシュートをしようと考えたが、守備側のある選手が非常に良い立ち位置にいたため他のプレーを選択せざるを得なかった」となれば、その守備選手の好プレーであったと考えるのが自然だろう。或いは、「ある攻撃側の選手が囮の動きをした結果、シュートコースが空いたため、味方のプレーヤーが得点を決めることができた」というのもサッカーでは良くあるプレーだ。しかしこれらのファインプレーを指標化することは（先のxGのような指標と比較して）難しい。なぜなら現在のプレースコアの基本はボールを保持しているプレーヤーを中心に考えているものだからである。

StatsBomb 360 Data はオープンデータとして有用だが致命的な欠点がある

この傾向はサッカーのプレーデータを見ても分かるところである。サッカーのプレーデータで有名なのはOpta, Wyscoutなどがあるが、これらは商用であり、一般にアクセスできるものの中で最も細かいデータが取得できるのは StatsBomb が公開しているオープンデータだろう。細かい説明は他に任せるが、1試合を通じてのボールタッチに関するほぼ全てのデータが取得することができる。例えばこちらに書かれている手順で、あるシーンのボールタッチ・シュートを下記のように可視化することが可能である。

また StatsBomb が画期的なのは、最近のプレーデータについてはボールにアクションしたプレーヤーだけでなく、ボール非保持の選手の位置を（恐らく中継画像を解析して）まとめた 360 Data というデータを同梱していることにある。これによって、原理的にはある選手がボール非保持のときにどのように効果的な（あるいは非効果的な）プレーができているのかを判別することができるようになる。これによって「質の良い動き」のできている選手、チームを判定することができるようになり、プレーの解析・市場価値の算出などに応用できる。

ただし、「理論的には」と書いたのには理由があり、Stats Bomb 360 Data はボール非保持の選手の座標を持っているのだが、それがどのプレーヤーであるかは識別できないようになっている。つまり、ある時刻フレームにおいてとある場所にいる選手が誰であるのかが分からないだけでなく、その前後のフレームにおいてその選手がどのような動きをしていたのかも、データから読み取ることはできない。

これは、試合の中継画像を解析する際に選手を同定する要素（背番号など）を常には取得できなかったことに原因があるのではないかと思われる。しかしこれは StatsBomb に特有の問題というわけではなく、世界のほとんどの試合においては、選手が高精度GPSを身につけてプレーしていないことを考えると、多くの試合において、存在している中継画像（動画）から個別の選手位置を同定する仕組みを考えることは有用である。

前置きが長くなってしまったが、以下のセクションではその手法をテクニカルに解説する。

StatsBomb 360 Data において Player ID を推定する方法

推定の流れ

まず、選手の座標（観測座標）データに対してPlayer IDを推定する流れを説明する。

最初に、StatsBombに特有の座標系を分析者にとって分かりやすいものへと変換する工程を踏む
その後、時刻フレーム（＝試合中のイベント（パス、ドリブル、チャージ等）ごとに記録されるスナップショット）間における観測座標間の距離の算出、中継画像の可視領域である Visible Area と観測座標との距離の算出を行う。これにより、ある座標の選手が次のフレームで移動する可能性のある座標に対し、それぞれいくらの距離を持っているかを計算できる
次に、これらのフレーム間の座標の対応をとる課題を、線形計画法の輸送問題として定式化する。数値計算で解を求めることにより、座標の対応を取得する
３で得られた対応に従い、全てのフレームを往復することで前後での座標をとっていく。このとき、StatsBombデータに記録されているボール保持選手についてのPlayer IDを、前後のフレームに予測Player IDとして伝搬させていく

以上によって、StatsBomb 360 Data には記載されていなかった選手のPlayer ID を推定する。なお、これらの処理はPythonによって記述し実施した。

Events データと 360 データの読み込み、座標系の変換

まず、StatsBombから取得したオープンデータをPandas の DataFrame として読み込む。データは試合（game_id）ごとにファイル格納されている。

import pandas

df_events      = pd.read_json(f"../open-data/data/events/{game_id}.json")
df_three_sixty = pd.read_json(f"../open-data/data/three-sixty/{game_id}.json")

次に、360データをEventsデータに紐付ける。このとき、座標系をSPADL形式（＝上記で可視化したプレーデータに準ずる形式）になるように変換する。StatsBombの座標系は、ホーム・アウェイ関係なくプレイヤーが所属するチームは右側に向かって攻撃するように定義されている（更にプレーの種別によって例外がある）。これでは攻守が逆転した瞬間に選手の座標系が反転してしまうという問題があるので、SPADL形式に準じてチームの攻撃方向を固定して見られるようにする。

import math

def convert_xy(x, y, inversion=False):
    x = min(120, max(1, x if x else 1))
    y = min(80, max(1, y if y else 1))
    x = ((x - 1) / 119) * 105.0
    y = 68 - ((y - 1) / 79) * 68.0
    if inversion:
        x = 105.0 - x
        y = 68.0 - y
    return x, y

def judge_home_team_player(freeze_frame, is_home_team):
    if is_home_team:
        if freeze_frame['teammate'] is True:
            return True
        else:
            return False
    else:
        if freeze_frame['teammate'] is True:
            return False
        else:
            return True

def select_color(freeze_frame, is_home_team, is_home_team_player=None):
    if is_home_team_player is None:
        is_home_team_player = judge_home_team_player(freeze_frame, is_home_team)
    if is_home_team_player:
        if freeze_frame['actor'] is True:
            return "red"
        elif freeze_frame['keeper'] is True:
            return "darkorange"
        else:
            return "orange"
    else:
        if freeze_frame['actor'] is True:
            return "blue"
        elif freeze_frame['keeper'] is True:
            return "darkblue"
        else:
            return "lightblue"

def judge_home_team(home_team_id, team_id, df_events, _event_id):
    is_home_team = (team_id == home_team_id)
    
    ball_receipt = df_events[df_events["id"] == _event_id]["ball_receipt"].tolist()[0]
    if isinstance(ball_receipt, dict) and ball_receipt['outcome']['id'] == 9: # Incomplete
        is_home_team = not is_home_team 
        
    dribble = df_events[df_events["id"] == _event_id]["dribble"].tolist()[0]
    if isinstance(dribble, dict) and dribble['outcome']['id'] == 9: # Incomplete
        is_home_team = not is_home_team
                
    return is_home_team

def judge_foul(df_events, _event_id):
    event_type = df_events[df_events["id"] == _event_id]["type"].tolist()[0]
    if isinstance(event_type, dict) and event_type['id'] == 21: # Foul Won
        return True
    else:
        return False

location_data = []

for index, row in df_events.iterrows():

    event_id          = row["id"]
    event_timestamp   = row["timestamp"]
    related_event_ids = row["related_events"]
    idx               = row["index"]
    team_id           = row["team"]['id']

    is_home_team = judge_home_team(home_team_id, team_id, df_events, event_id)
    is_foul      = judge_foul(df_events, event_id)
    inversion = is_home_team if is_foul else not is_home_team

    obj_freeze_frames = df_three_sixty[df_three_sixty["event_uuid"] == event_id]["freeze_frame"]
    obj_visible_area  = df_three_sixty[df_three_sixty["event_uuid"] == event_id]["visible_area"]

    if len(obj_freeze_frames) == 0:
        continue

    if not isinstance(row["player"], dict) or "id" not in row["player"]:
        continue
        
    player_id = row["player"]["id"]

    for freeze_frames in obj_freeze_frames:
        for ff in freeze_frames:
            is_home_team_player = judge_home_team_player(ff, is_home_team)
            pid = player_id if ff["actor"] is True else None
            locations.append({"teammate": ff["teammate"], "actor": ff["actor"], "keeper": ff["keeper"], "location": ff["location"], 
                              "xy": [x, y], "home_team_player": is_home_team_player, "player_id": pid, 'player_id_expected': None})

    location_data.append({"event_id": event_id, "locations": locations, "visible_area": visible_area, "timestamp": event_timestamp})

フレーム間における観測座標間の距離の算出、Visible Area との距離の算出

次に、location_data に格納されたフレームごとのデータから、各フレームで観測された座標（＝選手）の組み合わせに対して距離を計算する。また、Visible Area（画面領域）外に選手が出た場合を考えてこれを定義する線分との距離を計算する。

from scipy.spatial import distance
from numpy.linalg import norm

FIELD_SIZE_X, FIELD_SIZE_Y = 105.0, 68.0
FIELD_SIZE_XY = math.sqrt(pow(FIELD_SIZE_X, 2) + pow(FIELD_SIZE_Y, 2))

def split_locations_home_away(locations):
    locations_home, locations_home_keeper, locations_away, locations_away_keeper = [], [], [], []
    for location in locations:
        if location["home_team_player"]:
            if location["keeper"]:
                locations_home_keeper.append(location)
            else:
                locations_home.append(location)
        else:
            if location["keeper"]:
                locations_away_keeper.append(location)
            else:
                locations_away.append(location)
    return locations_home, locations_home_keeper, locations_away, locations_away_keeper

def calc_distance_matrix(locations_1, locations_2):
    xys_1 = [location["xy"] for location in locations_1]
    xys_2 = [location["xy"] for location in locations_2]
    return distance.cdist(xys_1, xys_2, metric='euclidean')

def calc_distance_and_neighbor_point(a, b, p):
    ap = p - a
    ab = b - a
    ba = a - b
    bp = p - b
    if np.dot(ap, ab) < 0:
        distance = norm(ap)
        neighbor_point = a
    elif np.dot(bp, ba) < 0:
        distance = norm(p - b)
        neighbor_point = b
    else:
        ai_norm = np.dot(ap, ab)/norm(ab)
        neighbor_point = a + (ab)/norm(ab)*ai_norm
        distance = norm(p - neighbor_point)
    return (neighbor_point, distance)

def is_touchline(va1, va2):
    if va1[0] == 0.0 and va2[0] == 0.0:
        return True
    elif va1[0] == FIELD_SIZE_X and va2[0] == FIELD_SIZE_X:
        return True
    elif va1[1] == 0.0 and va2[1] == 0.0:
        return True
    elif va1[1] == FIELD_SIZE_Y and va2[1] == FIELD_SIZE_Y:
        return True
    else:
        return False

def _calc_distances_location_visible_area(xys, visible_area):
    distances = np.zeros((len(xys), 1), dtype=np.float64)
    for i, xy in enumerate(xys):
        _distances = np.zeros((len(visible_area) - 1, 1), dtype=np.float64)
        for j in range(0, len(visible_area)-1):
            va1, va2 = visible_area[j], visible_area[j+1]
            if is_touchline(va1, va2):
                d = FIELD_SIZE_XY
            else:
                _, d = calc_distance_and_neighbor_point(np.asarray(va1), np.asarray(va2), np.asarray(xy))
            _distances[j, 0] = d
        distances[i, 0] = np.min(_distances)
    return distances

def calc_distances_location_visible_area(locations, visible_area):
    xys = [location["xy"] for location in locations]
    return _calc_distances_location_visible_area(xys, visible_area)

def calc_distances_location_visible_area_xy(locations, visible_area_x, visible_area_y):
    xys = [location["xy"] for location in locations]
    distances = np.zeros((len(xys), 2), dtype=np.float64)
    distances[:, 0:1] = _calc_distances_location_visible_area(xys, visible_area_x)
    distances[:, 1:2] = _calc_distances_location_visible_area(xys, visible_area_y)
    return distances.max(axis=1).reshape((len(xys), 1))

def make_lp_matrix(xy_distance_matrix, x_distances, y_distances):
    lp_matrix = np.zeros((x_distances.shape[0]+1, y_distances.shape[0]+1), dtype=np.float64)
    lp_matrix[0:x_distances.shape[0], 0:y_distances.shape[0]] = xy_distance_matrix
    lp_matrix[0:x_distances.shape[0], y_distances.shape[0]:y_distances.shape[0]+1] = x_distances
    lp_matrix[y_distances.shape[0]:y_distances.shape[0]+1, 0:y_distances.shape[0]] = y_distances.T
    return lp_matrix

def filter_unreal_distance(lp_matrix, s, meter_per_second=10.0, s_noise=0.5):
    return np.where(lp_matrix > meter_per_second * (s + s_noise), FIELD_SIZE_XY, lp_matrix)

線形計画法（LP：Linear Programming）としての定式化

フレーム間の点同士の距離から点の対応をとる課題を、線形計画法の輸送問題として定式化する。輸送問題とは、供給地（warehouses）にある物資（supply）を、需要地（projects）の需要（demand）を満たすように運ぶ最適解を求める問題である。ただし供給地から需要地までの経路にはコスト（costs）が存在し、その総コストを最小にしなければならない。Pythonで輸送問題を解く方法としてはこちらのページを参考にした。

点同士の対応問題に応用すると、最初のフレームの点（座標）を供給地、次のフレームの点（座標）を需要地と考え、各供給地から需要地まで量が1の物資を運ぶ問題として考える。需要地から供給地までのコストは点同士の距離が相当する。画面外はひとつの領域として考え、画面内の座標から画面外への移動、その逆は可視領域を構成する線分に対する距離を移動することに相当する。

from pulp import *
from collections import defaultdict

def make_assignments(lp_matrix, verbose=False, number_of_assignments=10):
    
    num_x, num_y = lp_matrix.shape
    assignments_y_x, assignments_x_y = {}, {}
    
    warehouses = [str(i) for i in range(num_x)]
    projects   = [str(i) for i in range(num_y)]
    supply     = {str(i): 1 if i != (num_x - 1) else (number_of_assignments + 1 - num_x) for i in range(num_x)}
    demand     = {str(i): 1 if i != (num_y - 1) else (number_of_assignments + 1 - num_y) for i in range(num_y)}
    costs      = lp_matrix
    
    costs = makeDict([warehouses, projects], costs, 0)
    prob = LpProblem("Material_supply_Problem", LpMinimize)
    Routes = [(w, b) for w in warehouses for b in projects]
    vars = LpVariable.dicts("Route", (warehouses, projects), 0, None, LpInteger)
    
    prob += (
        lpSum([vars[w][b] * costs[w][b] for (w, b) in Routes]),
        "Sum_of_Transporting_Costs",
    )
    for w in warehouses:
        prob += (
            lpSum([vars[w][b] for b in projects]) == supply[w],
            "Sum_of_Products_out_of_warehouses_%s" % w,
        )
    for b in projects:
        prob += (
            lpSum([vars[w][b] for w in warehouses]) <= demand[b],
            "Sum_of_Products_into_projects_%s" % b,
        )
    for w in warehouses:
        for b in projects:
            if w == str(num_x - 1) and b == str(num_y - 1):
                continue
            prob += (
                vars[w][b] <= 1,
                "Sum_of_Products_on_routes_%s_%s" % (w, b),
            )
        
    prob.solve()
    if verbose:
        print("Status =", LpStatus[prob.status])
    
    for v in prob.variables():
        if v.varValue > 0:
            _, x_idx, y_idx = v.name.split("_")
            assignments_y_x[y_idx] = {x_idx: {"value": v.varValue}}
            assignments_x_y[x_idx] = {y_idx: {"value": v.varValue}}
        
    if verbose:
        print("Value of Objective Function =", value(prob.objective))
    
    return assignments_y_x, assignments_x_y, value(prob.objective)

import datetime

def get_distance(assignments_y_x, lp_matrix):
    for y in assignments_y_x:
        for x in assignments_y_x[y]:
            assignments_y_x[y][x]["distance"] = lp_matrix[int(x)][int(y)]
    return assignments_y_x

list_average_vof_home, list_assignments_y_x_home, list_assignments_x_y_home = [], [], []
list_average_vof_away, list_assignments_y_x_away, list_assignments_x_y_away = [], [], []

for i in range(0, len(location_data)-1):
    ldx = location_data[i]
    ldy = location_data[i+1]

    lcx_home, lcx_home_keeper, lcx_away, lcx_away_keeper = split_locations_home_away(ldx["locations"])
    lcy_home, lcy_home_keeper, lcy_away, lcy_away_keeper = split_locations_home_away(ldy["locations"])
    diff = float((ldy["timestamp"] - ldx["timestamp"]).total_seconds())
    
    # home team
    if len(lcx_home) == 0 or len(lcy_home) == 0:
        list_assignments_y_x_home.append({})
        list_assignments_x_y_home.append({})
        list_average_vof_home.append(FIELD_SIZE_XY)
    else:
        xy_distance_matrix_home = calc_distance_matrix(lcx_home, lcy_home)
        x_distances_home = calc_distances_location_visible_area_xy(lcx_home, ldx["visible_area"], ldy["visible_area"])
        y_distances_home = calc_distances_location_visible_area_xy(lcy_home, ldx["visible_area"], ldy["visible_area"])
        lp_matrix_home = make_lp_matrix(xy_distance_matrix_home, x_distances_home, y_distances_home)
        lp_matrix_home = filter_unreal_distance(lp_matrix_home, diff)

        assignments_y_x_home, assignments_x_y_home, vof = make_assignments(lp_matrix_home, verbose=False)
        assignments_y_x_home = get_distance(assignments_y_x_home, lp_matrix_home)
        assignments_x_y_home = get_distance(assignments_x_y_home, lp_matrix_home.T)
        list_assignments_y_x_home.append(assignments_y_x_home)
        list_assignments_x_y_home.append(assignments_x_y_home)
        list_average_vof_home.append(vof / len(assignments_x_y_home))

    # away team
    if len(lcx_away) == 0 or len(lcy_away) == 0:
        list_assignments_y_x_away.append({})
        list_assignments_x_y_away.append({})
        list_average_vof_away.append(FIELD_SIZE_XY)
    else:
        xy_distance_matrix_away = calc_distance_matrix(lcx_away, lcy_away)
        x_distances_away = calc_distances_location_visible_area_xy(lcx_away, ldx["visible_area"], ldy["visible_area"])
        y_distances_away = calc_distances_location_visible_area_xy(lcy_away, ldx["visible_area"], ldy["visible_area"])
        lp_matrix_away = make_lp_matrix(xy_distance_matrix_away, x_distances_away, y_distances_away)
        lp_matrix_away = filter_unreal_distance(lp_matrix_away, diff)

        assignments_y_x_away, assignments_x_y_away, vof = make_assignments(lp_matrix_away, verbose=False)
        assignments_y_x_away = get_distance(assignments_y_x_away, lp_matrix_away)
        assignments_x_y_away = get_distance(assignments_x_y_away, lp_matrix_away.T)
        list_assignments_y_x_away.append(assignments_y_x_away)
        list_assignments_x_y_away.append(assignments_x_y_away)
        list_average_vof_away.append(vof / len(assignments_x_y_away))

LP解からの Player ID 推定法

最後にLP解（assignments）から Player IDを推定する。ロジックが煩雑になるため細かくはコードを参照してもらいたいが、大まかに言えば、

当該フレームと参照フレームを考えたとき、当該フレームのある点に対応する参照フレームの点でPlayer IDが定義されていれば、当該フレームにおいてもその点を予測Player ID（player_id_expected）とする
それ以外において、参照フレームで予測Player IDが存在していれば、（当該フレームの別の点がそのPlayer IDでない限り）当該フレームの対応する点は、その予想Player IDを引き継ぐ

といった仕組みになっている。これを試合開始フレームから試合終了フレームまで前後に伝搬させることで、分かっているPlayer IDから予測Player IDを増やしていく、という仕組みである。ただし、そもそもフレーム間の点の対応が悪い（≒中継画像に拡大カメラなどが挟まるなどして、連続した画面でなかった）場合、これらの伝搬をリセットする。

def get_player_id_expected(locations_y, locations_x, assignments_y_x, is_reverse=False):
    player_id_y = list(filter(None, [x['player_id'] for x in locations_y]))
    player_id_y = player_id_y[0] if len(player_id_y) != 0 else -1
    
    for j in range(len(locations_y)):
        if str(j) in assignments_y_x and len(assignments_y_x[str(j)]) == 1:
            for i in range(len(locations_x)):
                if str(i) in assignments_y_x[str(j)]:
                    player_id_x          = locations_x[i]['player_id']
                    player_id_expected_x = locations_x[i]['player_id_expected']
                    
                    if not is_reverse:
                        if player_id_x is not None: # 参照座標にplayer_idが存在していた場合、そのままplayer_id_expectedとする
                            locations_y[j]['player_id_expected'] = player_id_x
                        elif player_id_expected_x is not None:
                            if player_id_expected_x != player_id_y: # 参照座標のplayer_id_expectedが当該フレームのplayer_idと一致しない場合、そのままplayer_id_expectedとする
                                locations_y[j]['player_id_expected'] = player_id_expected_x
                            else: # 参照座標のplayer_id_expectedが当該フレームのplayer_idと一致した場合        
                                if locations_y[j]['player_id'] == player_id_y: # それが当該座標のplayer_idであればplayer_id_expectedとする
                                    locations_y[j]['player_id_expected'] = player_id_expected_x
                                else: # そうでなければ、当該フレームの別座標にあるはずなので、当該座標のplayer_id_expectedはNoneとする
                                    locations_y[j]['player_id_expected'] = None
                        else:
                            pass # player_id_expected_x is None
                    else:
                        if player_id_x is not None: # 参照座標にplayer_idが存在していた場合、そのままplayer_id_expectedとする
                            locations_y[j]['player_id_expected'] = player_id_x
                        elif player_id_expected_x is not None:
                            if player_id_expected_x != player_id_y: # 参照座標のplayer_id_expectedが当該フレームのplayer_idと一致しない場合、そのままplayer_id_expectedとする
                                locations_y[j]['player_id_expected'] = player_id_expected_x
                            else:
                                # 逆順の場合、参照座標のplayer_id_expectedは当該フレームのplayer_idから推定したものなので、
                                # それを当該座標のplayer_id_expectedと逆推定してはいけない
                                pass
                                #locations_y[j]['player_id_expected'] = locations_y[j]['player_id_expected']
                        else:
                            pass

    return locations_y

def get_number_of_collect_player_id(locations):
    num_collect, num_incollect = 0, 0
    player_id = list(filter(None, [x['player_id'] for x in locations]))
    if len(player_id) == 0:
        return num_collect, num_incollect
    else:
        player_id = player_id[0]
        for location in locations:
            if location['player_id_expected'] == player_id: # not None
                if location['player_id'] == player_id:
                    num_collect += 1
                else:
                    num_incollect += 1
        return num_collect, num_incollect

def get_number_of_player_id_expected(locations):
    return len(list(filter(None, [x['player_id_expected'] for x in locations])))

N = len(_location_data) - 1
average_vof_away_threshold = 10.0
_location_data = copy.deepcopy(location_data)

for i in range(0, N): # Expect forward
    
    ldx = _location_data[i]
    ldy = _location_data[i+1]
    lcx_home, lcx_home_keeper, lcx_away, lcx_away_keeper = split_locations_home_away(ldx["locations"])
    lcy_home, lcy_home_keeper, lcy_away, lcy_away_keeper = split_locations_home_away(ldy["locations"])
    assignments_y_x_home = list_assignments_y_x_home[i]
    assignments_y_x_away = list_assignments_y_x_away[i]
    
    if list_average_vof_home[i] > average_vof_away_threshold or list_average_vof_away[i] > average_vof_away_threshold:
        continue
    
    lcy_home = get_player_id_expected(lcy_home, lcx_home, assignments_y_x_home)
    lcy_away = get_player_id_expected(lcy_away, lcx_away, assignments_y_x_away)        
    _location_data[i+1]["locations"] = lcy_home + lcy_home_keeper + lcy_away + lcy_away_keeper 

for i in range(N-1, -1, -1): # Expect backward
    
    ldx = _location_data[i]
    ldy = _location_data[i+1]
    lcx_home, lcx_home_keeper, lcx_away, lcx_away_keeper = split_locations_home_away(ldx["locations"])
    lcy_home, lcy_home_keeper, lcy_away, lcy_away_keeper = split_locations_home_away(ldy["locations"])
    assignments_x_y_home = list_assignments_x_y_home[i]
    assignments_x_y_away = list_assignments_x_y_away[i]

    if list_average_vof_home[i] > average_vof_away_threshold or list_average_vof_away[i] > average_vof_away_threshold:
        continue
    
    lcx_home = get_player_id_expected(lcx_home, lcy_home, assignments_x_y_home, is_reverse=True)
    lcx_away = get_player_id_expected(lcx_away, lcy_away, assignments_x_y_away, is_reverse=True)
    _location_data[i]["locations"] = lcx_home + lcx_home_keeper + lcx_away + lcx_away_keeper

パラメータと定量評価について

以上のようにして推定したPlayer IDの精度を定量的に評価する。前述のように、分かっているPlayer IDを前後のフレームへと伝搬させていくため、伝搬先での真のPlayer IDと予測Player IDとの一致度を計測することでこの予測方法の精度を測ることができる。

本来連続していない（繋げて考えてはいけない）フレーム間についての対応を無理やり取ろうとすると、伝搬先での予測精度が落ちることが予想される。一方で、このことを恐れて予測の伝搬度合を保守的に取りすぎると、予測したPlayer IDの数自体が少なくなってしまい、これでは本来の目的にそぐわない。従って、各フレームにおいて、点の数に対する真のPlayer IDと予測Player IDとで埋められる占有率を計測することで、予測効率を計測することができる。前者の予測精度と予測効率はトレードオフの関係にあるため、これを見ながらフレーム間の伝搬度合をコントロールするパラメータの値を考える。

import statistics
    
sum_collect, sum_incollect = 0, 0
occupancies = []

for i in range(0, N):
    
    ldx = _location_data[i]
    lcx_home, lcx_home_keeper, lcx_away, lcx_away_keeper = split_locations_home_away(ldx["locations"])
        
    num_collect_home, num_incollect_home = get_number_of_collect_player_id(lcx_home)
    num_player_id_expected_home          = get_number_of_player_id_expected(lcx_home)
    num_collect_away, num_incollect_away = get_number_of_collect_player_id(lcx_away)
    num_player_id_expected_away          = get_number_of_player_id_expected(lcx_away)
    sum_collect   += num_collect_home + num_collect_away
    sum_incollect += num_incollect_home + num_incollect_away
    if len(lcx_home) != 0:
        occupancies.append(num_player_id_expected_home / len(lcx_home))
    if len(lcx_away) != 0:
        occupancies.append(num_player_id_expected_away / len(lcx_away))
    
print(average_vof_away_threshold, sum_collect, sum_incollect, sum_collect / (sum_collect + sum_incollect), statistics.mean(occupancies))

このようにパラメータ（フレーム間の選手移動距離の平均に対する閾値）を変化させたときの、Player IDの推定精度と推定率（画面に見えている選手のうち推定できたものの割合）をある試合を通じて計測しプロットした。画面間で選手が1.0m以上移動したとみなしたときに推定を打ち切った場合、その精度は100%だったが推定率は3.6%にとどまった。一方で、この移動距離の閾値を20mにまで緩和した場合、推定精度は96.7%に落ちた一方で、平均的な推定率が40.2%に達した。以降の処理はこのパラメータを20mにして実施した。

推定率が100%にならない理由としては、サッカー中継の可視領域外に複数の選手が抜けた場合、戻ってきた選手が誰なのかが分からなくなってしまう点、また中継における選手の拡大によって俯瞰画像が常には継続していないという点があげられる。ただしボール周辺の選手についてはより高確率で推定されるため、試合の分析において重要な範囲は一定程度カバーできるのではないかと考えられる。

推定 Player ID　の例

このようにして推定したPlayer IDの例を図示する。例に上げるのは2023年8月19日の Bayer Leverkusen と RB Leipzig との試合におけるあるフレームの選手座標である。StatsBomb OpenDataにおけるgame_idは 3895052 である。

まず360 Dataのみから分かる情報を図示する。ホームチーム、アウェイチームの立ち位置が分かるという点で有用だが、Player IDの推定がなかった場合、ホームチーム、アウェイチーム合わせた（可視領域にいる）フィールドプレイヤー19人のうち、その位置にいるのが誰であるのかが分かっているのは1人のみである。

次に、上記の方法によってプレイヤーの推定を行った結果を図示する。このフレームにおいて、画面の可視領域に写っている19人の選手のうち15人に対して推定プレイヤー名（精度97%）を付与することができた。このような情報強化をすることによって、選手ポジショニングの定量化などより深い分析に繋げることができると考えられる。

推定 Player ID 付与済みの StatsBomb 360 データの掲載

最後に、今回解説した方法を使い、Bundesliga 2023/2024シーズンのレバークーゼンのゲーム、および Euro 2024 のゲームデータに対して推定Player IDを付与したデータをGiｔHubに公開した。試合分析の参考になればと思う。リクエストがあれば他のデータに対する追加処理やコードの整理などを行いたい。