FRC 2026シーズン以降のChampionship出場権について

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

“""

Championship (Regional-only teams) vs Regional Events — EPA Comparison

– Events / TeamYears は Statbotics Python パッケージを使用

– team_events だけ REST API を使用（/v3/team_events?event=…）、limit=1000でページング

– Others（比較対象）は “regional" のみ（district / district_cmp は完全除外）

– Championship 側は “Regionalチームのみ" かつ **下位 p%**（既定 10%）で評価

– Regionalチーム定義は 2モード：

strict : その年に district 無し（所属ベース）【既定】

eligible : strict ∪ 「その年に一度でも regional に出場したチーム」（出場実績ベース）

– 比較指標は avg / median / min / sd を選択可能（既定: avg）

“""

from __future__ import annotations

import argparse

import json

import math

import sys

import time

from typing import Any, Dict, List, Optional, Tuple, Iterable, Set

import numpy as np

import pandas as pd

import requests

from cachecontrol import CacheControl # type: ignore

from tqdm import tqdm

from statbotics import Statbotics

BASE_URL = “https://api.statbotics.io/v3"

PAGE_SIZE = 1000 # RESTページサイズ

# —————————–

# ユーティリティ：型・タイプ正規化

# —————————–

def normalize_event_type(raw: Any) -> str:

s=str(raw).strip().lower()

fora, bin [(“-“, “_"), (" “, “_")]:

s=s.replace(a, b)

aliases= {

“rgional": “regional",

“regionals": “regional",

“dist": “district",

“dcmp": “district_cmp",

“district_champ": “district_cmp",

“districtcmp": “district_cmp",

“cmp_div": “champs_div",

“cmpdiv": “champs_div",

“champsdiv": “champs_div",

“cmp": “champs",

“champ": “champs",

“championship": “champs",

“champsionship": “champs",

“einsten": “einstein",

}

returnaliases.get(s, s)

CHAMP_TYPES_FULL = {“champs", “champs_div", “cmp", “cmp_div", “einstein"}

CHAMP_TYPES_NO_EINSTEIN = {“champs", “champs_div", “cmp", “cmp_div"}

def is_championship_type(t_norm: str, include_einstein: bool) -> bool:

returnt_normin (CHAMP_TYPES_FULLifinclude_einsteinelseCHAMP_TYPES_NO_EINSTEIN)

# —————————–

# REST (/team_events のみ)

# —————————–

def make_session() -> requests.Session:

s=CacheControl(requests.Session())

s.headers.update({“User-Agent": “champ-epa-regional-only/1.1"})

returns

def api_get_json(sess: requests.Session, path: str, params: Optional[Dict[str, Any]] = None, retries: int = 3) -> Any:

url=BASE_URL+path

backoff=1.0

forattemptinrange(retries):

try:

r=sess.get(url, params=params, timeout=30)

ifr.status_code==200:

returnr.json()

ifr.status_codein (429, 502, 503, 504):

time.sleep(backoff); backoff*=2

continue

r.raise_for_status()

exceptException:

ifattempt<retries-1:

time.sleep(backoff); backoff*=2

continue

raise

raiseRuntimeError(f"GET failed after retries: {url} params={params}")

def fetch_team_events_for_event(sess: requests.Session, event_code: str, page_size: int = PAGE_SIZE) -> List[Dict[str, Any]]:

rows: List[Dict[str, Any]] = []

offset=0

whileTrue:

data=api_get_json(sess, “/team_events", {“event": event_code, “limit": page_size, “offset": offset})

ifnotisinstance(data, list):

raiseRuntimeError(f"Unexpected /team_events response for event={event_code}")

rows.extend(data)

iflen(data) <page_size:

break

offset+=page_size

returnrows

# —————————–

# EPA & team id 抽出

# —————————–

def pick_epa_from_team_event(row: Dict[str, Any]) -> Optional[float]:

epa=row.get(“epa")

ifisinstance(epa, dict):

tp=epa.get(“total_points")

ifisinstance(tp, dict):

v=tp.get(“mean")

ifisinstance(v, (int, float)): returnfloat(v)

br=epa.get(“breakdown")

ifisinstance(br, dict):

v=br.get(“total_points")

ifisinstance(v, (int, float)): returnfloat(v)

v=epa.get(“mean")

ifisinstance(v, (int, float)): returnfloat(v)

forkin (“total", “total_epa", “epa_total"):

v=epa.get(k)

ifisinstance(v, (int, float)): returnfloat(v)

forkin (“epa_total_points", “epa_mean", “epa_total"):

v=row.get(k)

ifisinstance(v, (int, float)): returnfloat(v)

ifisinstance(epa, (int, float)): returnfloat(epa)

returnNone

def get_team_id_from_row(row: Dict[str, Any]) -> Optional[int]:

if"team"inrow:

try:

returnint(row[“team"])

exceptException:

pass

key=row.get(“team_key")

ifisinstance(key, str) andkey.lower().startswith(“frc"):

try:

returnint(key[3:])

exceptException:

returnNone

returnNone

# —————————–

# 統計ユーティリティ

# —————————–

def percentile_rank(values: np.ndarray, target: float) -> float:

returnfloat((values<=target).mean() *100.0)

def cohens_d_single_vs_dist(single_val: float, dist_vals: np.ndarray) -> float:

x1=np.array([single_val], dtype=float)

x2=np.asarray(dist_vals, dtype=float)

n2=len(x2)

s2=np.std(x2, ddof=1) ifn2>1else0.0

ifn2<=1ors2==0.0: returnfloat(“nan")

return (x1.mean() -x2.mean()) /s2# 単一点 vs 分布 → zに一致

def bootstrap_pvalue_greater(other_vals: np.ndarray, champ_val: float, n_boot: int = 5000, seed: int = 0) -> float:

rng=np.random.default_rng(seed)

x=np.asarray(other_vals, dtype=float)

y=np.array([champ_val], dtype=float)

diff_obs=y.mean() -x.mean()

pool=np.concatenate([x, y])

n_x, n_y=len(x), len(y)

diffs= []

for_inrange(n_boot):

x_b=rng.choice(pool, size=n_x, replace=True)

y_b=rng.choice(pool, size=n_y, replace=True)

diffs.append(y_b.mean() -x_b.mean())

diffs=np.array(diffs)

returnfloat((diffs>=diff_obs).mean())

# —————————–

# チーム所属：Regional集合の構築

# —————————–

def is_non_district_team(year_row: Dict[str, Any]) -> bool:

“""

team_years 1行から 'districtに所属していない’ を判定。

代表的キー: 'district’（None または " なら非district）

フォールバック: 'district_key’ / 'district_name’ などが空/Noneでも非district扱い

“""

forkin (“district", “district_key", “district_name"):

ifkinyear_row:

v=year_row.get(k)

ifvisNone: returnTrue

ifisinstance(v, str) andv.strip() =="": returnTrue

returnFalse

# 何もキーがないときは保守的に False

returnFalse

def make_events_dataframe(sb: Statbotics, year: int) -> pd.DataFrame:

events: List[Dict[str, Any]] = []

offset=0

whileTrue:

batch=sb.get_events(year=year, limit=PAGE_SIZE, offset=offset)

ifnotisinstance(batch, list) orlen(batch) ==0:

break

events.extend(batch)

iflen(batch) <PAGE_SIZE:

break

offset+=PAGE_SIZE

ifnotevents:

raiseRuntimeError(f"No events found for {year}")

df=pd.DataFrame(events)

if"event"notindf.columns:

forcandin (“key", “event_key", “code"):

ifcandindf.columns:

df=df.rename(columns={cand: “event"})

break

if"event"notindf.columns:

raiseRuntimeError(“イベントコード列(event/key/code)が見つかりません")

df[“__type_norm"] =df[“type"].map(normalize_event_type)

returndf

def build_regional_team_sets(sb: Statbotics, year: int, regional_events: List[str], sess: requests.Session) -> Tuple[Set[int], Set[int]]:

“""

戻り値:

strict_set : 非district所属チーム（所属ベース）

played_regio : その年に一度でも regional に出場したチーム（出場実績ベース）

“""

# strict_set：team_years から構築

strict_set: Set[int] =set()

offset=0

whileTrue:

batch=sb.get_team_years(year=year, limit=PAGE_SIZE, offset=offset)

ifnotisinstance(batch, list) orlen(batch) ==0:

break

forrowinbatch:

try:

team=int(row.get(“team"))

exceptException:

continue

ifis_non_district_team(row):

strict_set.add(team)

iflen(batch) <PAGE_SIZE:

break

offset+=PAGE_SIZE

# played_regio：regional の team_events を総なめして収集

played_regio: Set[int] =set()

forevintqdm(regional_events, desc="Collecting teams that played regional (for eligible mode)"):

try:

te_rows=fetch_team_events_for_event(sess, ev, page_size=PAGE_SIZE)

exceptExceptionase:

print(f"[WARN] {ev}: /team_events failed while collecting regional teams: {e}", file=sys.stderr)

continue

forrowinte_rows:

tid=get_team_id_from_row(row)

iftidisnotNone:

played_regio.add(tid)

returnstrict_set, played_regio

# —————————–

# 下位 p% 抽出ヘルパ

# —————————–

def bottom_percent_values(vals: np.ndarray, percent: float) -> np.ndarray:

“""配列 vals を昇順ソートして下位 p% を返す（少なくとも1件、端数は切り捨て）。"""

ifvals.size==0:

returnvals

k=int(np.floor(vals.size* (percent/100.0)))

k=max(1, k)

idx=np.argsort(vals, kind="mergesort") # 安定ソート

returnvals[idx][:k]

# —————————–

# メイン

# —————————–

def main():

ap=argparse.ArgumentParser(description="Championship (Regional-only teams) vs Regional events — EPA comparison (Champ bottom p%)")

ap.add_argument(“–year", type=int, default=2025, help="Season year (default: 2025)")

ap.add_argument(“–exclude-einstein", action="store_true", help="Exclude Einstein from Championship set")

ap.add_argument(“–metric", type=str, default="avg", choices=[“avg", “median", “min", “sd"], help="Comparison metric per event")

ap.add_argument(“–regional-mode", type=str, default="strict", choices=[“strict", “eligible"],

help="Regional team definition: strict(non-district only) or eligible(strict ∪ played regional). Default=strict")

ap.add_argument(“–champ-bottom-percent", type=float, default=10.0,

help="Championship は下位 p%% のチームのみで評価（既定: 10）")

ap.add_argument(“–out-prefix", type=str, default="", help="Output filename prefix (optional)")

args=ap.parse_args()

year=args.year

include_einstein=notargs.exclude_einstein

metric=args.metric

regional_mode=args.regional_mode # 'strict’ (default) or 'eligible’

champ_bottom_p=float(args.champ_bottom_percent)

prefix= (args.out_prefix +"_") ifargs.out_prefix else""

sb=Statbotics()

sess=make_session()

# 1) イベント一覧（Regional と Championship を抽出）

ev_df=make_events_dataframe(sb, year)

regional_events=ev_df.loc[ev_df[“__type_norm"] =="regional", “event"].astype(str).tolist()

champ_df_meta=ev_df[ev_df[“__type_norm"].apply(lambdat: is_championship_type(t, include_einstein))].copy()

ifchamp_df_meta.empty:

raiseRuntimeError(“No Championship events detected with current settings (Einstein excluded? check).")

champ_events=champ_df_meta[“event"].astype(str).tolist()

# 2) Regionalチーム集合（strict / eligible）

strict_set, played_regio=build_regional_team_sets(sb, year, regional_events, sess)

allowed_teams: Set[int] =strict_setifregional_mode=="strict"else (strict_set|played_regio)

# 3) Regional 各イベントの統計（allowed_teams 全体を使用）

per_event_rows: List[Dict[str, Any]] = []

forev_codeintqdm(regional_events, desc="Processing regional events"):

try:

te_rows=fetch_team_events_for_event(sess, ev_code, page_size=PAGE_SIZE)

exceptExceptionase:

print(f"[WARN] {ev_code}: /team_events failed: {e}", file=sys.stderr)

continue

epa_vals_all: List[float] = []

forrowinte_rows:

tid=get_team_id_from_row(row)

iftidisNoneortidnotinallowed_teams:

continue

v=pick_epa_from_team_event(row)

ifvisnotNoneandnp.isfinite(v):

epa_vals_all.append(float(v))

iflen(epa_vals_all) >=3:

arr_all=np.array(epa_vals_all, dtype=float)

n_used=arr_all.size# regional側は全量使用

per_event_rows.append({

“event": ev_code,

“event_name": ev_df.loc[ev_df[“event"] ==ev_code, “name"].values[0] if"name"inev_df.columnselseev_code,

“type": “regional",

“is_championship": False,

“n_teams_total_filtered": int(arr_all.size),

“n_teams_used": int(n_used),

“avg_team_epa": float(np.mean(arr_all)),

“sd_team_epa": float(np.std(arr_all, ddof=1)) ifarr_all.size>1elsefloat(“nan"),

“median_team_epa": float(np.median(arr_all)),

“min_team_epa": float(np.min(arr_all)),

})

else:

print(f"[WARN] {ev_code}: insufficient filtered teams (n={len(epa_vals_all)})", file=sys.stderr)

# 4) Championship（Regional-only teams）→ **下位 p%** だけでイベント統計

forev_codeintqdm(champ_events, desc=f"Processing championship events (regional teams; bottom {champ_bottom_p:.1f}%)"):

try:

te_rows=fetch_team_events_for_event(sess, ev_code, page_size=PAGE_SIZE)

exceptExceptionase:

print(f"[WARN] {ev_code}: /team_events failed: {e}", file=sys.stderr)

continue

epa_vals_all: List[float] = []

forrowinte_rows:

tid=get_team_id_from_row(row)

iftidisNoneortidnotinallowed_teams:

continue

v=pick_epa_from_team_event(row)

ifvisnotNoneandnp.isfinite(v):

epa_vals_all.append(float(v))

# まず Regional-only でフィルタした総数を確認

iflen(epa_vals_all) >=3:

arr_all=np.array(epa_vals_all, dtype=float)

# 下位 p% を抽出

arr_bot=bottom_percent_values(arr_all, champ_bottom_p)

ifarr_bot.size>=3:

per_event_rows.append({

“event": ev_code,

“event_name": ev_df.loc[ev_df[“event"] ==ev_code, “name"].values[0] if"name"inev_df.columnselseev_code,

“type": “championship",

“is_championship": True,

“n_teams_total_filtered": int(arr_all.size),

“n_teams_used": int(arr_bot.size), # ★下位p%の採用数

“avg_team_epa": float(np.mean(arr_bot)),

“sd_team_epa": float(np.std(arr_bot, ddof=1)) ifarr_bot.size>1elsefloat(“nan"),

“median_team_epa": float(np.median(arr_bot)),

“min_team_epa": float(np.min(arr_bot)),

})

else:

print(f"[WARN] {ev_code}: bottom {champ_bottom_p}% too small after filtering (n={arr_bot.size})", file=sys.stderr)

else:

print(f"[WARN] {ev_code}: insufficient filtered teams (n={len(epa_vals_all)})", file=sys.stderr)

ifnotper_event_rows:

raiseRuntimeError(“No events with sufficient filtered teams to compute stats.")

per_event_df=pd.DataFrame(per_event_rows)

# 5) 指標の選択と集約定義

metric_col= {

“avg": “avg_team_epa",

“median": “median_team_epa",

“min": “min_team_epa",

“sd": “sd_team_epa",

}[metric]

defchamp_aggregate(df: pd.DataFrame) -> float:

ifdf.empty: returnfloat(“nan")

ifmetric=="avg":

returnfloat(np.average(df[metric_col].values, weights=df[“n_teams_used"].values))

elifmetric=="sd":

vals=df[metric_col].values

ns=df[“n_teams_used"].values

weights=np.maximum(ns-1, 1) # (n-1)重み

returnfloat(np.average(vals, weights=weights))

else:

# median / min はイベント値の単純平均

returnfloat(np.mean(df[metric_col].values))

# 6) CSV 出力（イベントごとの統計）

out_csv= (f"{prefix}event_epa_averages_{year}_regional_only_mode-{regional_mode}_"

f"{metric}_champ-bottom-{int(champ_bottom_p)}.csv")

per_event_df.sort_values(

[“is_championship", metric_col, “n_teams_used"], ascending=[False, False, False]

).to_csv(out_csv, index=False)

# 7) 比較：Champ（Regional-only bottom p%） vs Regional（others）

champ_df=per_event_df[per_event_df[“is_championship"] ==True] # noqa: E712

others_df=per_event_df[per_event_df[“is_championship"] ==False] # noqa: E712

ifchamp_df.emptyorothers_df.empty:

raiseRuntimeError(“Not enough events in either Championship or Regional to compare.")

champ_val=champ_aggregate(champ_df)

other_vals=others_df[metric_col].values

mu=float(np.mean(other_vals))

sd=float(np.std(other_vals, ddof=1)) iflen(other_vals) >1elsefloat(“nan")

z= (champ_val-mu) /sdifsdandnp.isfinite(sd) andsd>0elsefloat(“nan")

d=cohens_d_single_vs_dist(champ_val, other_vals)

pct=percentile_rank(other_vals, champ_val)

p_boot=bootstrap_pvalue_greater(other_vals, champ_val)

summary= {

“year": year,

“comparison_metric": metric,

“regional_mode": regional_mode, # strict (default) / eligible

“include_einstein": include_einstein,

“champ_bottom_percent": champ_bottom_p,

“championship_value": champ_val,

“others_mean": mu,

“others_sd": sd,

“z_score_vs_others": z,

“cohens_d_vs_others": d,

“percentile_rank_vs_others": pct,

“bootstrap_p_one_sided": p_boot,

“n_events_championship": int(champ_df.shape[0]),

“n_events_regional": int(others_df.shape[0]),

“outputs": {“per_event_csv": out_csv},

}

out_json= (f"{prefix}championship_vs_regional_summary_{year}_mode-{regional_mode}_"

f"{metric}_champ-bottom-{int(champ_bottom_p)}.json")

withopen(out_json, “w", encoding="utf-8″) asf:

json.dump(summary, f, ensure_ascii=False, indent=2)

# 8) コンソール要約

print(f"\n=== Championship (Regional-only teams, bottom {champ_bottom_p:.1f}%) vs Regional — {year} ===")

print(f"Mode = {regional_mode} | Metric = {metric} | Einstein = {'included’ifinclude_einsteinelse’excluded’}")

print(f"Championship value (bottom {champ_bottom_p:.1f}%): {champ_val:.2f}")

print(f"Regional mean: {mu:.2f} (sd={sd:.2f}, n={len(other_vals)})")

print(f"Z-score (Champ vs Regional): {z:.2f}")

print(f"Cohen’s d: {d:.2f}")

print(f"Percentile of regional: {pct:.1f}%")

print(f"Bootstrap p (one-sided, Champ>Reg): {p_boot:.4f}")

print(f"\nSaved per-event table -> {out_csv}")

print(f"Saved summary JSON -> {out_json}")

if __name__ == “__main__":

try:

main()

exceptExceptionase:

print(f"\n[ERROR] {e}\n", file=sys.stderr)

print(“Tips:")

print(" – Regional集合は –regional-mode で切替（既定: strict）。")

print(" – Championship側は –champ-bottom-percent でサブセット率を変更できます。")

print(" – /team_events のEPAフィールドが合わない場合は pick_epa_from_team_event() を調整してください。")

sys.exit(1)