Steam同時接続（CCU）の自前収集と可視化【5分ポーリング／移動平均】Pythonテンプレ

この記事でできること：Steamの同時接続（CCU）を公式API（GetNumberOfCurrentPlayers）で定期収集し、時系列化→移動平均→可視化までを最小コストで構築します。複数タイトル対応、欠損時のリトライ、CSV/SQLite保存、日次集計とグラフ作成のテンプレを配布します。

全体像は Steamガイド（入口ページ）、前提のAppID取得は Step1: AppIDの見つけ方、公式APIの基本は Step2: 公式Web API を参照してください。

CCU収集の基本設計

データ源：ISteamUserStats/GetNumberOfCurrentPlayers（瞬間値＝スナップショット）。
頻度：5〜10分間隔が現実的（短すぎるとノイズ・負荷増）。
保存：最初はCSVが手軽。中長期はSQLiteで安定運用。
可視化：7日/14日等の移動平均で日周・週次ノイズを均す。
補助：履歴の参照だけなら外部可視化（SteamCharts/SteamDB）が便利。自前で分析するなら本記事の収集パイプを推奨。

クイックスタート：単発取得（確認用）

Python最小コード（1タイトルの現在CCU）

import requests

def get_ccu(appid:int) -> int | None:
    url = "https://api.steampowered.com/ISteamUserStats/GetNumberOfCurrentPlayers/v1/"
    r = requests.get(url, params={"appid": appid}, timeout=30).json()
    return (r.get("response") or {}).get("player_count")

print("CS2:", get_ccu(730))

AppIDが未確定の方は Step1 を先にどうぞ。

ポーリング収集テンプレ（CSV版）

複数タイトル×5分ポーリング→CSV追記（落ちても再開しやすい設計）

"""
config:
  APPIDS = 例: [570, 730, 1172470]  # Dota2, CS2, Apex など
  INTERVAL_SEC = 300  # 5分
"""
import os, time, csv, requests, datetime as dt
from pathlib import Path

APPIDS = [570, 730, 1172470]
INTERVAL_SEC = 300
OUT = Path("ccu_log.csv")

def utcnow_iso():
    return dt.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"

def get_ccu(appid:int) -> int | None:
    url = "https://api.steampowered.com/ISteamUserStats/GetNumberOfCurrentPlayers/v1/"
    for i in range(3):
        try:
            r = requests.get(url, params={"appid": appid}, timeout=20)
            r.raise_for_status()
            return (r.json().get("response") or {}).get("player_count")
        except Exception as e:
            if i == 2: 
                return None
            time.sleep(1.0 * (i+1))

def append_rows(rows:list[dict]):
    is_new = not OUT.exists()
    with OUT.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["ts_utc","appid","ccu"])
        if is_new:
            w.writeheader()
        for row in rows:
            w.writerow(row)

def one_loop():
    ts = utcnow_iso()
    rows = []
    for aid in APPIDS:
        c = get_ccu(aid)
        if c is not None:
            rows.append({"ts_utc": ts, "appid": aid, "ccu": c})
    if rows:
        append_rows(rows)
        print(ts, "rows:", len(rows))

if __name__ == "__main__":
    while True:
        one_loop()
        time.sleep(INTERVAL_SEC)

停止→再開してもCSV追記で継続できます。より確実にしたい場合はSQLite版を使うと重複排除や欠損補正が容易です。

SQLite版（重複排除・クエリが楽）

シンプルなスキーマ＋UPSERTで重複を自動排除

import time, sqlite3, requests, datetime as dt
from pathlib import Path

APPIDS = [570, 730, 1172470]
INTERVAL_SEC = 300
DB = Path("ccu.db")

def init_db():
    con = sqlite3.connect(DB)
    con.execute("""
        CREATE TABLE IF NOT EXISTS ccu(
          ts_utc TEXT NOT NULL,
          appid INTEGER NOT NULL,
          ccu INTEGER,
          PRIMARY KEY(ts_utc, appid)
        );
    """)
    con.commit()
    return con

def get_ccu(appid:int) -> int | None:
    url = "https://api.steampowered.com/ISteamUserStats/GetNumberOfCurrentPlayers/v1/"
    for i in range(3):
        try:
            r = requests.get(url, params={"appid": appid}, timeout=20)
            r.raise_for_status()
            return (r.json().get("response") or {}).get("player_count")
        except Exception:
            if i == 2: return None
            time.sleep(1.0 * (i+1))

def utcnow_iso():
    return dt.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"

def one_loop(con):
    ts = utcnow_iso()
    rows = []
    for aid in APPIDS:
        c = get_ccu(aid)
        if c is not None:
            rows.append((ts, aid, c))
    if rows:
        con.executemany("INSERT OR REPLACE INTO ccu(ts_utc, appid, ccu) VALUES (?,?,?)", rows)
        con.commit()
        print(ts, "rows:", len(rows))

if __name__ == "__main__":
    con = init_db()
    while True:
        one_loop(con)
        time.sleep(INTERVAL_SEC)

可視化：移動平均とピーク把握

CSV→日次集計→7日移動平均→PNG出力（matplotlib）

import pandas as pd
import matplotlib.pyplot as plt

df = pd.read_csv("ccu_log.csv", parse_dates=["ts_utc"])
df["date"] = df["ts_utc"].dt.tz_localize("UTC").dt.tz_convert("Asia/Tokyo").dt.date
daily = (df.groupby(["date","appid"])["ccu"]
           .mean().reset_index(name="ccu_mean"))
daily["ma7"] = daily.groupby("appid")["ccu_mean"].transform(lambda s: s.rolling(7, min_periods=1).mean())

for aid, g in daily.groupby("appid"):
    plt.figure(figsize=(8,4))
    g = g.sort_values("date")
    plt.plot(g["date"], g["ccu_mean"], label="Daily mean")
    plt.plot(g["date"], g["ma7"], label="7-day MA")
    plt.title(f"AppID {aid} - CCU trend")
    plt.xlabel("Date (JST)")
    plt.ylabel("CCU")
    plt.legend()
    plt.tight_layout()
    plt.savefig(f"ccu_{aid}.png", dpi=150)
    plt.close()

SQLite→直近30日を重ね描き（曜日パターンの確認）

import sqlite3, pandas as pd, matplotlib.pyplot as plt, datetime as dt

con = sqlite3.connect("ccu.db")
df = pd.read_sql_query("SELECT ts_utc, appid, ccu FROM ccu", con, parse_dates=["ts_utc"])
df["ts_jst"] = df["ts_utc"].dt.tz_localize("UTC").dt.tz_convert("Asia/Tokyo")
df["date"] = df["ts_jst"].dt.date
df["hour"] = df["ts_jst"].dt.hour
last_date = df["date"].max()
start = pd.to_datetime(last_date) - pd.Timedelta(days=30)
mask = pd.to_datetime(df["date"]) >= start

for aid, g in df[mask].groupby("appid"):
    piv = g.pivot_table(index="hour", columns="date", values="ccu", aggfunc="mean")
    plt.figure(figsize=(8,4))
    plt.plot(piv.index, piv, alpha=0.3)
    plt.title(f"AppID {aid} - Hourly CCU (last 30d)")
    plt.xlabel("Hour of Day (JST)")
    plt.ylabel("CCU")
    plt.tight_layout()
    plt.savefig(f"ccu_hourly_last30_{aid}.png", dpi=150)
    plt.close()

運用：スケジューラ設定（Windows/Mac/Linux）

Windows：「タスクスケジューラ」で5分おきに実行。引数に仮想環境のPythonとスクリプトのパス。
macOS/Linux：crontab -e → */5 * * * * /path/to/python /path/to/ccu_poll.py >> ccu.log 2>&1
監視：ログに成功件数を出す／Prometheus系にメトリクスを送ると異常検知が容易。