ポケモンカードのAIエージェント開発大会に参加する方法

cat > fetch_samples.sh <<'SH'
set -euo pipefail

mkdir -p data notebooks agents

echo "Kaggle CLI:"
kaggle --version
echo

echo "Downloading competition files..."
kaggle competitions download -c pokemon-tcg-ai-battle -p data -o
unzip -o data/pokemon-tcg-ai-battle.zip -d data || true

# Keep deterministic order.
NAMES=(
  sample_lucario
  sample_abomasnow
  sample_dragapult
  sample_iono
)

KERNELS=(
  kiyotah/a-sample-rule-based-agent-mega-lucario-ex-deck
  kiyotah/a-sample-rule-based-agent-mega-abomasnow-ex-deck
  kiyotah/a-sample-rule-based-agent-dragapult-ex-deck
  kiyotah/a-sample-rule-based-agent-iono-s-deck
)

for i in "${!NAMES[@]}"; do
  name="${NAMES[$i]}"
  kernel="${KERNELS[$i]}"

  echo "Fetching $name: $kernel"

  mkdir -p "notebooks/$name" "agents/$name"

  # Source notebook/script + metadata.
  kaggle kernels pull "$kernel" -p "notebooks/$name" -m

  # Output artifacts from the latest public run; ideally contains agent.py/deck.csv or an archive.
  kaggle kernels output "$kernel" -p "agents/$name" -o || true

  # Unpack common archive formats if present.
  find "agents/$name" -maxdepth 1 -type f \( -name '*.zip' -o -name '*.tar.gz' -o -name '*.tgz' \) -print | while read -r f; do
    case "$f" in
      *.zip) unzip -o "$f" -d "agents/$name" ;;
      *.tar.gz|*.tgz) tar -xzf "$f" -C "agents/$name" ;;
    esac
  done
done

echo
echo "Files found:"
find agents notebooks -maxdepth 3 -type f | sort

echo
echo "Agent/deck check:"
find agents -maxdepth 3 -type f \( -name 'agent.py' -o -name 'deck.csv' \) -print | sort
SH

bash fetch_samples.sh

import argparse
import csv
import importlib.util
import itertools
import multiprocessing as mp
import os
import sys
import time
from concurrent.futures import ProcessPoolExecutor, as_completed
from contextlib import contextmanager
from pathlib import Path

import pandas as pd
from kaggle_environments import make
from tqdm import tqdm


_SDK_READY = False


def ensure_sdk_path():
    """
    Make the local competition SDK importable.

    Sample agents import `cg.api`. Locally, `cg` may be in:
      - /work/data/sample_submission/cg
      - /work/agents/<agent_name>/cg
      - another extracted competition folder
    """
    global _SDK_READY

    if _SDK_READY:
        return

    candidates = []

    preferred = [
        Path("/work/data/sample_submission"),
        Path.cwd() / "data" / "sample_submission",
    ]

    for p in preferred:
        if (p / "cg").exists():
            candidates.append(p)

    # Fallback scan. This is done once per process.
    for root in [Path.cwd(), Path("/work")]:
        if not root.exists():
            continue
        for init_file in root.rglob("cg/__init__.py"):
            candidates.append(init_file.parent.parent)

    seen = set()

    # Put candidates on sys.path. Keep order stable and avoid duplicates.
    for p in candidates:
        p = p.resolve()
        if p in seen:
            continue
        seen.add(p)

        if (p / "cg").exists() and str(p) not in sys.path:
            sys.path.insert(0, str(p))

    try:
        import cg  # noqa: F401
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError(
            "Could not import `cg`. Check where the SDK is located with:\n"
            "  find /work -type d \\( -name 'cg-lib' -o -name 'cg' \\) -print\n"
            "Then make sure the parent directory of `cg/` is on PYTHONPATH."
        ) from e

    _SDK_READY = True


@contextmanager
def import_context(path: Path):
    """
    Temporarily import/run inside an agent directory.

    This matters because sample agents may depend on local files next to main.py.
    """
    ensure_sdk_path()

    path = path.resolve()
    old_cwd = Path.cwd()
    old_sys_path = list(sys.path)

    os.chdir(path)
    sys.path.insert(0, str(path))

    try:
        yield
    finally:
        os.chdir(old_cwd)
        sys.path[:] = old_sys_path


def read_deck(path: Path) -> list[int]:
    vals = []

    with path.open() as f:
        for row in csv.reader(f):
            if not row:
                continue

            token = row[0].strip()

            if token and token.lstrip("-").isdigit():
                vals.append(int(token))

    if len(vals) != 60:
        raise ValueError(f"{path} contains {len(vals)} numeric card IDs, expected 60.")

    return vals


def load_agent(agent_dir: Path, module_suffix: str):
    """
    Load an agent function from agents/<name>/main.py.

    Returns a wrapper that restores the agent directory context on every call.
    This protects agents that do imports or file reads inside agent(obs).
    """
    ensure_sdk_path()

    agent_dir = agent_dir.resolve()
    agent_py = agent_dir / "main.py"

    if not agent_py.exists():
        raise FileNotFoundError(f"Missing {agent_py}")

    module_name = f"agent_{agent_dir.name}_{module_suffix}_{time.time_ns()}"
    spec = importlib.util.spec_from_file_location(module_name, agent_py)
    mod = importlib.util.module_from_spec(spec)

    with import_context(agent_dir):
        assert spec.loader is not None
        spec.loader.exec_module(mod)

    if not hasattr(mod, "agent"):
        raise AttributeError(f"{agent_py} does not define agent(obs_dict).")

    base_agent = mod.agent

    def wrapped_agent(obs_dict):
        old_cwd = os.getcwd()
        path_str = str(agent_dir)
        added_path = False

        os.chdir(agent_dir)

        if not sys.path or sys.path[0] != path_str:
            sys.path.insert(0, path_str)
            added_path = True

        try:
            return base_agent(obs_dict)
        finally:
            if added_path:
                try:
                    sys.path.remove(path_str)
                except ValueError:
                    pass
            os.chdir(old_cwd)

    return wrapped_agent


def final_result(env):
    final = env.steps[-1]

    r0 = final[0].get("reward", 0)
    r1 = final[1].get("reward", 0)
    s0 = final[0].get("status")
    s1 = final[1].get("status")

    if r0 > r1:
        winner = 0
    elif r1 > r0:
        winner = 1
    else:
        winner = -1

    return winner, r0, r1, s0, s1


def run_game_loaded(
    agent0_dir: Path,
    agent1_dir: Path,
    deck0: list[int],
    deck1: list[int],
    agent0,
    agent1,
    game_id: int,
):
    """
    Run one game using already-loaded decks and agents.
    """
    ensure_sdk_path()

    started = time.time()
    error = None

    try:
        env = make("cabt", configuration={"decks": [deck0, deck1]}, debug=False)
        env.run([agent0, agent1])
        winner, reward0, reward1, status0, status1 = final_result(env)

    except Exception as e:
        winner, reward0, reward1, status0, status1 = -2, None, None, "ERROR", "ERROR"
        error = repr(e)

    elapsed = time.time() - started

    return {
        "game_id": game_id,
        "p0": agent0_dir.name,
        "p1": agent1_dir.name,
        "winner": winner,
        "winner_name": agent0_dir.name if winner == 0 else agent1_dir.name if winner == 1 else "draw_or_error",
        "reward0": reward0,
        "reward1": reward1,
        "status0": status0,
        "status1": status1,
        "seconds": elapsed,
        "error": error,
    }


def run_one(agent0_dir: Path, agent1_dir: Path, game_id: int):
    """
    Safe mode: reload decks and agents every game.
    Slower, but avoids cross-game global-state leakage.
    """
    ensure_sdk_path()

    started = time.time()
    error = None

    try:
        agent0_dir = agent0_dir.resolve()
        agent1_dir = agent1_dir.resolve()

        deck0 = read_deck(agent0_dir / "deck.csv")
        deck1 = read_deck(agent1_dir / "deck.csv")

        agent0 = load_agent(agent0_dir, f"p0_g{game_id}")
        agent1 = load_agent(agent1_dir, f"p1_g{game_id}")

        env = make("cabt", configuration={"decks": [deck0, deck1]}, debug=False)
        env.run([agent0, agent1])
        winner, reward0, reward1, status0, status1 = final_result(env)

    except Exception as e:
        winner, reward0, reward1, status0, status1 = -2, None, None, "ERROR", "ERROR"
        error = repr(e)

    elapsed = time.time() - started

    return {
        "game_id": game_id,
        "p0": agent0_dir.name,
        "p1": agent1_dir.name,
        "winner": winner,
        "winner_name": agent0_dir.name if winner == 0 else agent1_dir.name if winner == 1 else "draw_or_error",
        "reward0": reward0,
        "reward1": reward1,
        "status0": status0,
        "status1": status1,
        "seconds": elapsed,
        "error": error,
    }


def run_pair_batch(
    agent0_dir_str: str,
    agent1_dir_str: str,
    start_game_id: int,
    n_games: int,
    reuse_agents: bool,
):
    """
    Worker process entrypoint.

    Runs a chunk of games for one ordered matchup:
      agent0 as player 0, agent1 as player 1.

    If reuse_agents=True, each agent is imported once per chunk.
    This is faster but assumes no meaningful cross-game global state.
    """
    ensure_sdk_path()

    agent0_dir = Path(agent0_dir_str).resolve()
    agent1_dir = Path(agent1_dir_str).resolve()

    rows = []

    if reuse_agents:
        try:
            deck0 = read_deck(agent0_dir / "deck.csv")
            deck1 = read_deck(agent1_dir / "deck.csv")

            agent0 = load_agent(agent0_dir, f"p0_batch_{start_game_id}")
            agent1 = load_agent(agent1_dir, f"p1_batch_{start_game_id}")

        except Exception:
            # Fall back to per-game error handling so failures are recorded as rows.
            for i in range(n_games):
                rows.append(run_one(agent0_dir, agent1_dir, start_game_id + i))
            return rows

        for i in range(n_games):
            rows.append(
                run_game_loaded(
                    agent0_dir=agent0_dir,
                    agent1_dir=agent1_dir,
                    deck0=deck0,
                    deck1=deck1,
                    agent0=agent0,
                    agent1=agent1,
                    game_id=start_game_id + i,
                )
            )

    else:
        for i in range(n_games):
            rows.append(run_one(agent0_dir, agent1_dir, start_game_id + i))

    return rows


def rows_to_df(rows: list[dict]) -> pd.DataFrame:
    if not rows:
        return pd.DataFrame()

    df = pd.DataFrame(rows)

    if "game_id" in df.columns:
        df = df.sort_values("game_id").reset_index(drop=True)

    return df


def save_games_atomic(rows: list[dict], path: Path):
    """
    Write games.csv atomically so Ctrl+C during a write is unlikely to corrupt it.
    """
    df = rows_to_df(rows)
    tmp = path.with_suffix(".csv.tmp")
    df.to_csv(tmp, index=False)
    os.replace(tmp, path)


def summarize(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return pd.DataFrame(
            columns=[
                "agent",
                "games",
                "wins",
                "losses",
                "draws",
                "errors",
                "win_rate_ex_errors",
                "avg_seconds",
            ]
        )

    rows = []
    names = sorted(set(df["p0"]) | set(df["p1"]))

    for name in names:
        as_p0 = df[df["p0"] == name]
        as_p1 = df[df["p1"] == name]
        all_games = pd.concat([as_p0, as_p1], ignore_index=True)

        wins = ((as_p0["winner"] == 0).sum() + (as_p1["winner"] == 1).sum())
        losses = ((as_p0["winner"] == 1).sum() + (as_p1["winner"] == 0).sum())
        draws = (all_games["winner"] == -1).sum()
        errors = (all_games["winner"] == -2).sum()
        games = wins + losses + draws + errors

        rows.append(
            {
                "agent": name,
                "games": int(games),
                "wins": int(wins),
                "losses": int(losses),
                "draws": int(draws),
                "errors": int(errors),
                "win_rate_ex_errors": wins / max(1, wins + losses + draws),
                "avg_seconds": all_games["seconds"].dropna().mean(),
            }
        )

    return pd.DataFrame(rows).sort_values(
        ["win_rate_ex_errors", "wins"], ascending=False
    )


def load_existing_games(games_path: Path):
    if not games_path.exists():
        return [], 0, {}

    existing = pd.read_csv(games_path)

    if existing.empty:
        return [], 0, {}

    rows = existing.to_dict("records")
    game_id = int(existing["game_id"].max())
    completed_counts = existing.groupby(["p0", "p1"]).size().to_dict()

    return rows, game_id, completed_counts


def build_jobs(agent_dirs: list[Path], args, completed_counts: dict, start_game_id: int):
    """
    Build chunked ordered-matchup jobs.

    For 4 agents:
      permutations(agent_dirs, 2) => 12 ordered matchups.
    Each ordered matchup is split into chunks for better checkpointing.
    """
    jobs = []
    game_id = start_game_id

    schedule = list(itertools.permutations(agent_dirs, 2))

    for a0, a1 in schedule:
        key = (a0.name, a1.name)
        already_done = int(completed_counts.get(key, 0))
        remaining = max(0, args.games_per_seat - already_done)

        while remaining > 0:
            n_games = min(args.chunk_size, remaining)
            batch_start_game_id = game_id + 1
            game_id += n_games

            jobs.append(
                {
                    "agent0": str(a0.resolve()),
                    "agent1": str(a1.resolve()),
                    "start_game_id": batch_start_game_id,
                    "n_games": n_games,
                    "p0": a0.name,
                    "p1": a1.name,
                }
            )

            remaining -= n_games

    return jobs, game_id


def print_planned_jobs(jobs: list[dict]):
    if not jobs:
        print("No pending jobs.")
        return

    total_games = sum(j["n_games"] for j in jobs)
    print(f"Pending batches: {len(jobs)}")
    print(f"Pending games:   {total_games}")

    preview = pd.DataFrame(jobs)
    preview = preview.groupby(["p0", "p1"])["n_games"].sum().reset_index()

    print("\nPending games by ordered matchup:")
    print(preview.to_string(index=False))


def main():
    # Avoid BLAS/OpenMP oversubscription when using multiple worker processes.
    os.environ.setdefault("OMP_NUM_THREADS", "1")
    os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")
    os.environ.setdefault("MKL_NUM_THREADS", "1")
    os.environ.setdefault("NUMEXPR_NUM_THREADS", "1")

    ensure_sdk_path()

    ap = argparse.ArgumentParser()
    ap.add_argument("--agents-dir", default="agents")
    ap.add_argument("--games-per-seat", type=int, default=10)
    ap.add_argument("--out-dir", default="runs/baseline")
    ap.add_argument("--workers", type=int, default=max(1, (os.cpu_count() or 2) - 1))
    ap.add_argument(
        "--chunk-size",
        type=int,
        default=10,
        help="Number of games per worker batch. Larger is faster; smaller checkpoints more often.",
    )
    ap.add_argument(
        "--reuse-agents",
        action="store_true",
        help="Import each agent once per chunk. Faster, but assumes agents are stateless across games.",
    )
    ap.add_argument(
        "--no-resume",
        action="store_true",
        help="Ignore existing games.csv and start this out-dir from scratch.",
    )

    args = ap.parse_args()

    if args.games_per_seat <= 0:
        raise SystemExit("--games-per-seat must be positive")

    if args.workers <= 0:
        raise SystemExit("--workers must be positive")

    if args.chunk_size <= 0:
        raise SystemExit("--chunk-size must be positive")

    agents_root = Path(args.agents_dir)
    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    games_path = out_dir / "games.csv"
    summary_path = out_dir / "summary.csv"
    matrix_path = out_dir / "p0_winrate_matrix.csv"

    agent_dirs = sorted(
        [
            p.resolve()
            for p in agents_root.iterdir()
            if p.is_dir() and (p / "main.py").exists() and (p / "deck.csv").exists()
        ]
    )

    if len(agent_dirs) < 2:
        raise SystemExit(
            f"Need at least two agent dirs with main.py and deck.csv under {agents_root}"
        )

    print("Agents:")
    for p in agent_dirs:
        print(" -", p.name)

    if args.no_resume:
        rows = []
        game_id = 0
        completed_counts = {}
        print("\nStarting fresh because --no-resume was set.")
    else:
        rows, game_id, completed_counts = load_existing_games(games_path)
        if rows:
            print(f"\nResuming from {len(rows)} completed games.")
        else:
            print("\nNo existing games.csv found. Starting fresh.")

    jobs, last_assigned_game_id = build_jobs(
        agent_dirs=agent_dirs,
        args=args,
        completed_counts=completed_counts,
        start_game_id=game_id,
    )

    print()
    print_planned_jobs(jobs)

    print()
    print(f"Workers:      {args.workers}")
    print(f"Chunk size:   {args.chunk_size}")
    print(f"Reuse agents: {args.reuse_agents}")

    if jobs:
        ctx = mp.get_context("fork" if sys.platform.startswith("linux") else "spawn")

        try:
            with ProcessPoolExecutor(
                max_workers=args.workers,
                mp_context=ctx,
            ) as ex:
                futures = [
                    ex.submit(
                        run_pair_batch,
                        job["agent0"],
                        job["agent1"],
                        job["start_game_id"],
                        job["n_games"],
                        args.reuse_agents,
                    )
                    for job in jobs
                ]

                for fut in tqdm(
                    as_completed(futures),
                    total=len(futures),
                    desc="matchup batches",
                ):
                    batch_rows = fut.result()
                    rows.extend(batch_rows)

                    # Atomic checkpoint after each completed chunk.
                    save_games_atomic(rows, games_path)

        except KeyboardInterrupt:
            print("\nInterrupted. Saving completed batches.")
            save_games_atomic(rows, games_path)
            raise

    df = rows_to_df(rows)

    if not df.empty:
        save_games_atomic(rows, games_path)

    summary = summarize(df)
    summary.to_csv(summary_path, index=False)

    print("\nSummary:")
    print(summary.to_string(index=False))

    if not df.empty:
        matrix_source = df[df["winner"].isin([0, 1])]

        if not matrix_source.empty:
            matrix = matrix_source.pivot_table(
                index="p0",
                columns="p1",
                values="winner",
                aggfunc=lambda s: (s == 0).mean(),
            )

            matrix.to_csv(matrix_path)

            print("\nP0 win-rate matrix:")
            print(matrix.round(3).to_string())
        else:
            print("\nNo non-draw, non-error games available for p0 win-rate matrix.")

    print()
    print(f"Wrote: {games_path}")
    print(f"Wrote: {summary_path}")

    if matrix_path.exists():
        print(f"Wrote: {matrix_path}")


if __name__ == "__main__":
    main()