Steamの現在同時接続数（CCU）を1回取得して保存する方法と、保存CSV・架空CSVから観測の記録を可視化する方法をまとめます。
自分のデータに合う実行経路を選ぶ
以下はWindows PowerShellでの手順です。Python 3.10以上を用意し、必要なファイルを保存したフォルダーをエクスプローラーで開き、アドレス欄に powershell と入力してEnterを押します。Get-Location で作業フォルダーを確認してください。コード全文は後ろに掲載しています。選んだ経路だけを実行し、別の経路のコードを同じファイルへつなげないでください。
CCUは取得瞬間の同時接続人数です。下の3経路から必要なものだけ選びます。練習のためにタスクスケジューラや定期収集を始める必要はありません。
現在値をAPIで1回取得・保存する
「CCU取得の考え方」の関数コード→「CSVに追記する」のコードだけを順に collect_ccu.py へ保存します。末尾のAppID（配布例730）を確認します。実行1回につき1回取得し、APIキーは不要です。
python -m pip install requests
python collect_ccu.pycollect_ccu.pyは自分のファイルの場所を基準に outputs/steam_ccu.csv へ追記します。この取得コードは正常時に保存先を表示しません。スクリプト横のCSVを開き、末尾のtimestamp_utc・appid・player_countを確認します。ほかの分析例の相対パスは作業フォルダー基準なので、スクリプトのあるフォルダーから実行して揃えます。
既存CCU CSVを集計する（API通信なし）
visualize_steam_ccu.pyだけを保存します。入力は outputs/steam_ccu.csv。取得間隔に応じて引数を指定します。
python -m pip install pandas plotly
python visualize_steam_ccu.py --interval-minutes 60 --tolerance-minutes 15架空CSVで記録漏れと実測0を試す（API通信なし）
make_steam_samples.pyと本記事のvisualize_steam_ccu.pyを同じフォルダーへ保存し、pandas・Plotlyを導入します。
python make_steam_samples.py
python visualize_steam_ccu.py --input outputs/fictional_steam_sample/steam_ccu.csv --output outputs/fictional_ccu_charts出力は既存CSV経路が outputs/steam_ccu_charts/、架空経路が outputs/fictional_ccu_charts/。ccu.html で線が途切れる時間帯、excluded_rows.csv のreasonで除外行を確認します。CSV自体にない未観測時刻の一覧を作るわけではありません。画面とvalidation.jsonの件数を照合します。同じ出力先への再実行は集計を更新します。
架空AppID 9000001は1時の実測0を残し、1時→5時は線を分けます。1月1日は4観測・観測平均9.25・観測最大15です。daily_ccu.csvのobserved_meanは時間平均ではなく、observed_maxも取り逃しのない日中ピークではありません。保存済み集計だけを読む別の描画経路はありません。
掲載コードと詳しい条件
CCU取得の考え方
Steam Web APIの ISteamUserStats/GetNumberOfCurrentPlayers を使うと、指定したAppIDの現在同時接続数を取得できます。この値は「取得した瞬間の値」です。過去推移を見たい場合は、自分で定期取得して保存する必要があります。
import requests
def get_current_players(appid: int) -> int:
url = "https://api.steampowered.com/ISteamUserStats/GetNumberOfCurrentPlayers/v1/"
params = {"appid": appid}
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()["response"]
return int(data["player_count"])CSVに追記する
import csv
from datetime import datetime, timezone
from pathlib import Path
def append_ccu(appid: int, csv_path: Path) -> None:
player_count = get_current_players(appid)
now = datetime.now(timezone.utc).isoformat()
csv_path.parent.mkdir(parents=True, exist_ok=True)
fields = ["timestamp_utc", "appid", "player_count"]
has_header = csv_path.exists() and csv_path.stat().st_size > 0
if has_header:
with csv_path.open(encoding="utf-8-sig", newline="") as existing:
if next(csv.reader(existing), None) != fields:
raise ValueError("CSVの列名が異なります。既存ファイルを確認してください")
with csv_path.open("a", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["timestamp_utc", "appid", "player_count"])
if not has_header:
writer.writeheader()
writer.writerow({
"timestamp_utc": now,
"appid": appid,
"player_count": player_count,
})
if __name__ == "__main__":
append_ccu(730, Path(__file__).resolve().parent / "outputs/steam_ccu.csv")定期実行する
最初の関数定義とCSV追記コードを順に collect_ccu.py へ保存します。python -m pip install requests pandas plotly の後、保存フォルダーで python collect_ccu.py を実行します。CSVはスクリプトと同じフォルダー内の outputs/steam_ccu.csv です。タスクスケジューラにはPython実行ファイルの絶対パス、引数にこのスクリプトの絶対パス、開始フォルダーに保存フォルダーを指定してください。同じCSVへの同時実行は避けます。可視化コードは取得とは別に、同じ保存フォルダーから実行します。
Windowsならタスクスケジューラ、Linuxならcronで定期実行できます。まずは1時間に1回など、負荷の低い間隔で十分です。細かすぎる間隔で取得しても、分析に使い切れないことが多いです。
保存CSVから、記録漏れを区切って集計・可視化する
上の取得・追記処理とは別のPython実行です。collect_ccu.py と同じ作業フォルダーに次のコードを visualize_steam_ccu.py として保存してください。過去の記録を新たに取得する処理でも、定期実行タスクを作る処理でもありません。
import argparse
import json
from pathlib import Path
import pandas as pd
import plotly.graph_objects as go
def analyze(path, output, interval_minutes=60, tolerance_minutes=15):
if interval_minutes <= 0 or tolerance_minutes < 0:
raise ValueError('取得間隔は正、許容幅は0以上にしてください')
df = pd.read_csv(path, dtype=str, keep_default_na=False)
required = {'timestamp_utc','appid','player_count'}
if missing := required - set(df.columns):
raise ValueError(f'必要列不足: {sorted(missing)}。取得CSVの保存先を確認してください')
df['source_row'] = range(2, len(df)+2)
df['time'] = pd.to_datetime(df.timestamp_utc, utc=True, errors='coerce', format='mixed')
df['count'] = pd.to_numeric(df.player_count, errors='coerce')
reasons = []
for r in df.itertuples():
bad = []
if not str(r.appid).isdigit(): bad.append('invalid_appid')
# UTC列にタイムゾーンのない文字列は黙ってUTCと推定しない。
if pd.isna(r.time) or not (str(r.timestamp_utc).endswith('Z') or str(r.timestamp_utc).endswith('+00:00')):
bad.append('invalid_or_non_utc_time')
if pd.isna(r.count) or r.count == float('inf'): bad.append('non_numeric_or_infinite')
elif r.count < 0: bad.append('negative_count')
elif r.count % 1: bad.append('non_integer_count')
reasons.append('|'.join(bad))
df['reason'] = reasons
invalid = df.loc[df.reason != ''].copy()
valid = df.loc[df.reason == ''].copy()
key = ['appid','time']
conflict = valid.groupby(key)['count'].transform('nunique') > 1
conflicted = valid.loc[conflict].copy(); conflicted['reason'] = 'conflicting_same_timestamp'
valid = valid.loc[~conflict].copy()
duplicate = valid.duplicated(key, keep='first')
duplicates = valid.loc[duplicate].copy(); duplicates['reason'] = 'identical_duplicate'
valid = valid.loc[~duplicate].sort_values(key).copy()
valid['day_utc'] = valid.time.dt.strftime('%Y-%m-%d')
daily = valid.groupby(['appid','day_utc']).agg(observation_count=('count','size'),
observed_mean=('count','mean'), observed_max=('count','max')).reset_index()
# 日次完全性の判定ではなく、単純な少数観測表示。
daily['few_observations'] = daily.observation_count < 3
output = Path(output); output.mkdir(parents=True, exist_ok=True)
daily.to_csv(output/'daily_ccu.csv', index=False, encoding='utf-8-sig')
pd.concat([invalid,conflicted,duplicates]).to_csv(output/'excluded_rows.csv', index=False, encoding='utf-8-sig')
fig = go.Figure(); segments = 0
threshold = pd.Timedelta(minutes=interval_minutes+tolerance_minutes)
for appid, group in valid.groupby('appid', sort=True):
cut = group.time.diff().gt(threshold).cumsum()
for number, part in group.groupby(cut):
segments += 1
fig.add_trace(go.Scatter(x=part.time.dt.strftime('%Y-%m-%d %H:%M:%S').tolist(),
y=part['count'].tolist(), mode='lines+markers', connectgaps=False,
name=f'AppID {appid}', legendgroup=str(appid), showlegend=number==0))
fig.update_layout(title=dict(text='CCU観測値（UTC）', font=dict(size=16)),
xaxis_title='観測日時 UTC', yaxis_title='同時接続人数',
legend=dict(orientation='h', y=1.18), height=480, autosize=True,
margin=dict(l=50,r=15,t=95,b=100), xaxis=dict(tickangle=-30))
if valid.empty:
fig.add_annotation(text='有効な観測なし：除外理由・入力を確認', showarrow=False, xref='paper', yref='paper', x=.5,y=.5)
fig.write_html(output/'ccu.html', include_plotlyjs=True, full_html=True,
config={'responsive':True,'displaylogo':False})
report = dict(input_rows=len(df), valid_observations=len(valid), invalid_rows=len(invalid),
conflicting_rows=len(conflicted), identical_duplicates=len(duplicates),
segments=segments, interval_minutes=interval_minutes, tolerance_minutes=tolerance_minutes,
display_timezone='UTC', few_observations_rule='observation_count < 3')
(output/'validation.json').write_text(json.dumps(report, indent=2), encoding='utf-8')
print(json.dumps(report)); return daily, report
if __name__ == '__main__':
p = argparse.ArgumentParser()
p.add_argument('--input', default='outputs/steam_ccu.csv')
p.add_argument('--output', default='outputs/steam_ccu_charts')
p.add_argument('--interval-minutes', type=float, default=60)
p.add_argument('--tolerance-minutes', type=float, default=15)
a = p.parse_args(); analyze(a.input, a.output, a.interval_minutes, a.tolerance_minutes)
入力は outputs/steam_ccu.csv、必要列は timestamp_utc・appid・player_count。UTC保存・UTC表示で統一します。AppIDと時刻で整列し、同時刻・同人数の重複は1件にします。同時刻で人数が競合した行は、どちらが正しいか推定せず両方を除外します。不正日時、UTC表記なし、非数値、負数、小数人数は理由と元CSV行番号を excluded_rows.csv へ保存します。実測0は残します。
60分間隔・15分の許容幅なら、75分を超える時刻差で線を別の区間に分けます。取得頻度に合わせて引数を指定してください。欠損行がCSVになくても時刻差で区切り、未観測を0や直前値で補いません。競合・欠損の除外後にも時刻差を判定します。
出力は outputs/steam_ccu_charts/ccu.html、daily_ccu.csv、excluded_rows.csv、validation.json です。HTMLはPlotlyを含む単体ファイルで、ブラウザで開けます（数MBになります）。再実行すると同じ出力先の集計を更新します。入力CSVは変更しません。
observation_count は日別観測件数、observed_mean は観測値の単純平均、observed_max は観測最大値です。厳密な時間平均・公式の日平均・取り逃しのない真の最大値ではありません。few_observations は3件未満という目安で、完全に取得できた日の判定ではありません。記録のない日の人数は不明です。CCUからDAU（日間利用者数）・購入者数・売上は計算できません。
架空CSVで記録漏れと実測0を試す
実測データがない場合は、架空サンプル生成コードを保存して実行してください。レビューCSVとCCU CSVを outputs/fictional_steam_sample/ に作ります。以下は別の出力先を使うので実観測CSVには触れません。
以下は掲載コードで生成した固定の架空見本です。実際のゲームの人数ではありません。AppID 9000001の1時から5時は線が途切れ、1時の0人は点として残ります。1月1日は4観測・平均9.25・観測最大15、翌日は1観測です。読者のCSVがこのサイトの見本へ送信される仕組みではありません。
分析時の注意
- 取得間隔が粗いとピークを逃す
- タイムゾーンはUTCで保存し、表示時に変換する
- メンテナンスや大型アップデートで急変することがある
- 1時点のCCUだけで人気や売上を断定しない
まとめ
Steamの同時接続数は現在値として取得できます。推移を見たい場合は、取得時刻と一緒にCSVへ蓄積し、後から可視化します。レビュー分析やアップデート前後の反応を見る補助指標として使うと扱いやすいです。