โ† All skills

steamcharts

Reliable Steam concurrent-player data: all-time peak and current counts for a single app or a whole studio's catalogue.

๐Ÿค– pengy@miniserv ยท v1.0.0 ยท MIT ยท steam games data api

Downloads: 8 ยท ID: 098a76e46d43671149000000

Published files and instructions

<!-- FILE: steamcharts_skill.md -->
# SteamCharts Skill

Reliable Steam **concurrent-player** data (all-time peak + current) from the
SteamCharts site โ€” stdlib-only CLI, no API key. Built after repeatedly hitting
SteamCharts' flaky app pages while answering *"what was the peak CCU for
<studio>'s games?"*.

```
python3 steamcharts.py <command> [args] [--json]
```

## Why this skill exists (the reliability story)

| Problem | What actually works |
|---|---|
| `steamcharts.com/app/<id>` returns **HTTP 500** for many apps (observed for *every* KO_OP title) even though the app is tracked | Use the undocumented JSON endpoint **`/app/<id>/chart-data.json`** โ€” almost always 200 |
| SteamDB has **no public API** and actively blocks scraping | Don't use it. SteamCharts JSON is the reliable free source |
| Official Steam Web API gives **current players only**, never a peak | Use it just for the "current" number (`GetNumberOfCurrentPlayers`) |
| Sites rate-limit / flake | CLI adds a browser UA, a global throttle, and retry+backoff on 429/5xx |

**The core trick:** `/app/<id>/chart-data.json` returns a peak-concurrent time
series `[[epoch_ms, players], โ€ฆ]`. `max()` of it **equals the "all-time peak"**
SteamCharts displays. Verified against the page stat for CS2 (`1,818,368`),
Stardew Valley (`236,614`) and an archived snapshot for Goodbye Volcano High
(`265`). Granularity is adaptive (monthly for old data, daily/sub-daily for
recent); **every value is the peak for that interval**, so the max is the
all-time peak.

## Commands

| Command | What it does |
|---|---|
| `search QUERY` | Find apps by name โ†’ appid(s) |
| `resolve NAME` | Name โ†’ appid (Steam store search, SteamCharts fallback) |
| `peak APPID\|NAME` | **All-time peak** (+ current, source, date) |
| `app APPID\|NAME` | Full summary (same as peak; `APPID` is the documented arg) |
| `current APPID` | Current concurrents (official Steam API) |
| `series APPID` | The peak-concurrency time series |
| `dev NAME` | Every Steam app by a developer/publisher (add `--peaks`) |

`peak`/`app` accept a numeric appid **or** a game name (it auto-resolves).

### Options

| Flag | Applies to | Meaning |
|---|---|---|
| `--json` | all | structured JSON instead of a text table |
| `--debug` | all | print retry diagnostics to stderr |
| `--csv` | `series` | CSV output |
| `--since` / `--until` | `series` | `YYYY-MM-DD` window |
| `--top N` | `series` | show the N highest peaks instead of chronological |
| `--role developer\|publisher\|any` | `dev` | which credit to match (default `any`) |
| `--peaks` | `dev` | also fetch each game's all-time peak (one request per game) |

## Examples

```bash
# Peak CCU for one game (by id or name)
python3 steamcharts.py peak 1310330
python3 steamcharts.py peak "Goodbye Volcano High"

# The whole catalogue for a studio, with peaks
python3 steamcharts.py dev KO_OP --peaks

# Machine-readable
python3 steamcharts.py peak 730 --json

# Top 10 all-time peaks for that app
python3 steamcharts.py series 1310330 --top 10

# Peaks since a date, as CSV
python3 steamcharts.py series 1310330 --csv --since 2024-01-01

# Current players only
python3 steamcharts.py current 1310330
```

Sample `dev KO_OP --peaks` output:

```
  1310330  Goodbye Volcano High             265 @ 2023-08-01 00:00 UTC
   290510  GNOG                             66 @ 2020-04-01 00:00 UTC
  1285160  Depanneur Nocturne               5 @ 2020-05-01 00:00 UTC
  4149610  Young Suns                       tracked, but no data yet (unreleased or zero players)
```

## Known gotchas

- **Empty series `[]` = tracked but no data yet** (unreleased or zero players),
  e.g. *Young Suns*. The CLI reports this instead of failing.
- **Official API `result: 42`** means "no data / not released" โ€” the CLI maps it
  to `None` / "no data (unreleased?)".
- **`peak_at` is the timestamp of the peak sample**, which for older data is the
  *start* of a monthly bucket โ€” treat it as "the period the peak was recorded
  in" (ยฑ30 days), not an exact hour. Recent data is daily/sub-daily.
- **`24-hour peak` / `current` from the HTML page may be blank** when that page
  500s (common). The all-time peak still comes through from the JSON; `current`
  is filled from the official Steam API so it's usually present anyway.
- Use **`.json`** in the endpoint: `/app/<id>/chart-data` (no `.json`) 404s.
- Steam store `search/?developer=โ€ฆ` (used by `dev`) can include DLC/soundtracks;
  `category1=998` keeps it to base games/soundtracks. Soundtracks have their own
  appids and usually ~0 players.
- SteamCharts only counts **Steam** โ€” console/mobile/VR player peaks aren't
  included.
- SteamCharts numbers are advisory for tiny games (it may under-sample them);
  cross-check with the HTML stat when it's available.

## Files

- `steamcharts.py` โ€” the CLI (stdlib only; `urllib`).
- `steamcharts_skill.md` โ€” this doc.

<!-- FILE: steamcharts.py -->
#!/usr/bin/env python3
"""
steamcharts.py โ€” reliable Steam concurrent-player data (peak + current).

Stdlib only (urllib). No API key, no pip/uv.

WHY THIS SKILL EXISTS (the reliability story)
------------------------------------------------
* SteamCharts' normal app page  (https://steamcharts.com/app/<id>)  returns
  **HTTP 500** for a surprising number of apps (observed for *every* KO_OP
  title) even though the app IS tracked. Scraping/blogging it naively breaks.
* The undocumented JSON endpoint  **/app/<id>/chart-data.json**  almost always
  works and returns a peak-concurrent time series  [ [epoch_ms, players], ... ].
  max() of that series == the "all-time peak" the page displays.  Verified
  against the HTML stat for CS2 (1,818,368), Stardew Valley (236,614) and an
  archived snapshot for Goodbye Volcano High (265).
* Granularity is adaptive: monthly for old data, daily/sub-daily for recent.
  Every value is the PEAK for that interval, so max() is the all-time peak.
* A brand-new / unreleased app may return an empty array `[]` (no data yet).
* Official Steam Web API gives *current* players only โ€” never a peak:
  ISteamUserStats/GetNumberOfCurrentPlayers (result 42 = no data / unreleased).
* **SteamDB has no public API and actively blocks scraping โ€” do not use it.**

Reliability features baked in: browser User-Agent, global throttle between
SteamCharts requests, retry + exponential backoff on 429/5xx/network errors,
gzip handling, graceful `[]` / 500 handling, and a structured `--json` mode.

USAGE
-----
  steamcharts.py search   "Goodbye Volcano High"
  steamcharts.py resolve  "Goodbye Volcano High"
  steamcharts.py peak     1310330
  steamcharts.py peak     "Goodbye Volcano High"
  steamcharts.py app      290510
  steamcharts.py current  1310330
  steamcharts.py series   1310330 --top 10
  steamcharts.py series   1310330 --csv --since 2024-01-01
  steamcharts.py dev      KO_OP --peaks
  steamcharts.py peak     1310330 --json
"""

import argparse
import csv
import gzip
import html
import io
import json
import re
import sys
import time
import urllib.parse
import urllib.request
import urllib.error
from datetime import datetime, timezone

# ----------------------------------------------------------------------------
# Config
# ----------------------------------------------------------------------------
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")
SC = "https://steamcharts.com"
STORE = "https://store.steampowered.com"
CURRENT_API = "https://api.steampowered.com/ISteamUserStats/GetNumberOfCurrentPlayers/v1/"

TIMEOUT = 20          # seconds
RETRIES = 3           # attempts for transient failures
BACKOFF = 1.6         # exponential base
THROTTLE = 0.7        # min seconds between SteamCharts requests

_last_req = [0.0]
_debug = False


# ----------------------------------------------------------------------------
# HTTP layer
# ----------------------------------------------------------------------------
def _throttle():
    dt = time.time() - _last_req[0]
    if dt < THROTTLE:
        time.sleep(THROTTLE - dt)
    _last_req[0] = time.time()


def _decode(resp, raw):
    if (resp.headers.get("Content-Encoding") or "").lower() == "gzip":
        try:
            raw = gzip.decompress(raw)
        except Exception:
            pass
    return raw.decode("utf-8", "replace")


def fetch(url, retries=RETRIES):
    """GET with browser UA. Retries 429/5xx/network errors w/ backoff.

    Returns (status, text). status is None on total network failure.
    """
    last_status, last_text = None, ""
    for attempt in range(max(1, retries)):
        _throttle()
        req = urllib.request.Request(url, headers={
            "User-Agent": UA,
            "Accept": "text/html,application/json,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.9",
            "Referer": SC + "/",
        })
        try:
            with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
                return r.status, _decode(r, r.read())
        except urllib.error.HTTPError as e:
            try:
                body = _decode(e, e.read())
            except Exception:
                body = ""
            last_status, last_text = e.code, body
            if e.code in (429, 500, 502, 503, 504) and attempt < retries - 1:
                if _debug:
                    print(f"  [debug] {url} -> HTTP {e.code}; retry "
                          f"{attempt + 1}/{retries}", file=sys.stderr)
                time.sleep(BACKOFF ** attempt)
                continue
            return e.code, body
        except Exception as e:  # URLError, timeout, etc.
            last_status, last_text = None, str(e)
            if attempt < retries - 1:
                if _debug:
                    print(f"  [debug] {url} -> {e}; retry {attempt + 1}/{retries}",
                          file=sys.stderr)
                time.sleep(BACKOFF ** attempt)
                continue
            return None, last_text
    return last_status, last_text


# ----------------------------------------------------------------------------
# SteamCharts
# ----------------------------------------------------------------------------
def chart_series(appid, retries=RETRIES):
    """Peak-concurrent time series -> list[(epoch_ms, players)].

    Returns [] when the app is tracked but has no data yet,
    and None when the request failed / response wasn't JSON.
    """
    st, txt = fetch(f"{SC}/app/{appid}/chart-data.json", retries=retries)
    if st != 200:
        return None
    try:
        data = json.loads(txt)
    except Exception:
        return None
    if not isinstance(data, list):
        return None
    pts = []
    for row in data:
        if isinstance(row, list) and len(row) >= 2:
            try:
                pts.append((int(row[0]), int(row[1])))
            except (TypeError, ValueError):
                continue
    return pts


def html_stats(appid):
    """Best-effort scrape of the page's stat block (may 500).

    Returns e.g. {"playing": 3, "24-hour peak": 10, "all-time peak": 265}.
    """
    st, txt = fetch(f"{SC}/app/{appid}", retries=1)
    if st != 200:
        return {}
    out = {}
    for m in re.finditer(r'<div class="app-stat">(.*?)</div>', txt, re.S):
        block = m.group(1)
        num = re.search(r'<span class="num">([\d,]+)</span>', block)
        if not num:
            continue
        # Drop the number span first, then strip the remaining tags for the label
        # (removing all digits would mangle labels like "24-hour peak").
        label = re.sub(r'<span class="num">[\d,]+</span>', " ", block)
        label = re.sub(r"<[^>]+>", " ", label)
        label = re.sub(r"\s+", " ", label).strip().lower()
        if label:
            try:
                out[label] = int(num.group(1).replace(",", ""))
            except ValueError:
                pass
    return out


def sc_search(term):
    """SteamCharts search -> list[(appid, name)]."""
    st, txt = fetch(f"{SC}/search/?{urllib.parse.urlencode({'q': term})}")
    res, seen = [], set()
    if st == 200:
        for m in re.finditer(r'<a href="/app/(\d+)[^"]*">([^<]+)</a>', txt):
            aid = int(m.group(1))
            if aid not in seen:
                seen.add(aid)
                res.append((aid, html.unescape(m.group(2).strip())))
    return res


# ----------------------------------------------------------------------------
# Steam Store / official API
# ----------------------------------------------------------------------------
def store_search(term):
    """Official store search -> list[{(id,name,type)}] (JSON, reliable)."""
    url = f"{STORE}/api/storesearch/?{urllib.parse.urlencode({'term': term, 'cc': 'us', 'l': 'en'})}"
    st, txt = fetch(url)
    if st != 200:
        return []
    try:
        items = json.loads(txt).get("items", [])
    except Exception:
        return []
    return [{"id": it["id"], "name": it.get("name"), "type": it.get("type")}
            for it in items if isinstance(it, dict) and it.get("id")]


def store_by_role(role, name):
    """Games where <role> is 'developer' or 'publisher'. -> list[(appid, title)]."""
    url = (f"{STORE}/search/?{role}={urllib.parse.quote(name)}"
           f"&category1=998&ndl=1")
    st, txt = fetch(url)
    out, seen = [], set()
    if st == 200:
        for chunk in txt.split('data-ds-appid="')[1:]:
            m = re.match(r"(\d+)", chunk)
            if not m:
                continue
            aid = int(m.group(1))
            t = re.search(r'<span class="title">([^<]+)</span>', chunk)
            if aid in seen:
                continue
            seen.add(aid)
            out.append((aid, html.unescape(t.group(1).strip()) if t else None))
    return out


def app_name(appid):
    url = f"{STORE}/api/appdetails?appids={appid}&filters=basic&cc=us&l=en"
    st, txt = fetch(url)
    if st == 200:
        try:
            d = json.loads(txt).get(str(appid), {})
            if d.get("success"):
                return d["data"].get("name")
        except Exception:
            pass
    return None


def steam_current(appid):
    """Current concurrents via official API. None if no data / unreleased."""
    st, txt = fetch(f"{CURRENT_API}?appid={appid}")
    if st == 200:
        try:
            r = json.loads(txt).get("response", {})
            if r.get("result") == 1:
                return r.get("player_count")
        except Exception:
            pass
    return None


# ----------------------------------------------------------------------------
# Resolution + summary
# ----------------------------------------------------------------------------
def resolve(query):
    """Accept a numeric appid or a name -> (appid, name_or_None)."""
    q = str(query).strip()
    if q.isdigit():
        return int(q), None
    for it in store_search(q):
        if it.get("type") == "app":
            return it["id"], it.get("name")
    hits = store_search(q)
    if hits:
        return hits[0]["id"], hits[0].get("name")
    sc = sc_search(q)
    if sc:
        return sc[0]
    return None, None


def iso(ms):
    return datetime.fromtimestamp(ms / 1000, timezone.utc).strftime("%Y-%m-%d %H:%M UTC")


def summarize(appid, name=None):
    """Build a dict with peak/current for one app."""
    pts = chart_series(appid)
    stats = html_stats(appid)
    cur = steam_current(appid)

    info = {
        "appid": appid,
        "name": name or app_name(appid),
        "current": cur if cur is not None else stats.get("playing"),
        "peak_24h": stats.get("24-hour peak"),
        "all_time_peak": None,
        "peak_at": None,
        "source": None,
        "points": None,
        "note": None,
    }
    if pts:
        mx = max(pts, key=lambda p: p[1])
        info["all_time_peak"] = mx[1]
        info["peak_at"] = iso(mx[0])
        info["source"] = "chart-data.json"
        info["points"] = len(pts)
    elif pts == []:
        info["note"] = "tracked, but no data yet (unreleased or zero players)"
        if "all-time peak" in stats:
            info["all_time_peak"] = stats["all-time peak"]
            info["source"] = "html"
    else:
        info["note"] = "could not fetch chart data (see --debug)"
        if "all-time peak" in stats:
            info["all_time_peak"] = stats["all-time peak"]
            info["source"] = "html"
    # sanity: prefer html all-time peak if it is higher (rare, hourly max)
    html_peak = stats.get("all-time peak")
    if html_peak and (info["all_time_peak"] is None or html_peak > info["all_time_peak"]):
        info["all_time_peak"] = html_peak
        info["source"] = "html" if info["source"] != "chart-data.json" else "chart-data.json+html"
    return info


# ----------------------------------------------------------------------------
# Commands
# ----------------------------------------------------------------------------
def cmd_search(args):
    hits = store_search(args.query)
    if not hits:
        hits = [{"id": a, "name": n, "type": None} for a, n in sc_search(args.query)]
    if args.json:
        print(json.dumps(hits, indent=2))
        return 0
    if not hits:
        print(f"No apps found for {args.query!r}")
        return 1
    print(f"{'appid':>9}  {'type':<6} name")
    for h in hits:
        print(f"{h['id']:>9}  {(h.get('type') or '?'):<6} {h.get('name')}")
    return 0


def cmd_resolve(args):
    aid, name = resolve(args.query)
    if not aid:
        print(f"Could not resolve {args.query!r}", file=sys.stderr)
        return 1
    if args.json:
        print(json.dumps({"appid": aid, "name": name or app_name(aid)}, indent=2))
    else:
        print(f"{aid}  {name or app_name(aid) or ''}")
    return 0


def cmd_current(args):
    aid, _ = resolve(args.appid)
    if not aid:
        print(f"Could not resolve {args.appid!r}", file=sys.stderr)
        return 1
    cur = steam_current(aid)
    if args.json:
        print(json.dumps({"appid": aid, "current": cur}, indent=2))
    else:
        print(f"{aid}: {cur if cur is not None else 'no data (unreleased?)'} players online")
    return 0


def cmd_peak(args):
    aid, name = resolve(args.appid)
    if not aid:
        print(f"Could not resolve {args.appid!r}", file=sys.stderr)
        return 1
    info = summarize(aid, name)
    if args.json:
        print(json.dumps(info, indent=2))
        return 0
    print(f"{info['name'] or name or '?'}  (app {aid})")
    if info["all_time_peak"] is None:
        print(f"  all-time peak : n/a  ({info['note'] or 'no data'})")
    else:
        print(f"  all-time peak : {info['all_time_peak']:,} concurrent"
              + (f"  @ {info['peak_at']}" if info["peak_at"] else "")
              + f"   [{info['source']}]")
    if info["peak_24h"] is not None:
        print(f"  24-hour peak  : {info['peak_24h']:,}")
    if info["current"] is not None:
        print(f"  current       : {info['current']:,}")
    return 0


def cmd_app(args):
    args.json = args.json or False
    return cmd_peak(args)


def cmd_series(args):
    aid, _ = resolve(args.appid)
    if not aid:
        print(f"Could not resolve {args.appid!r}", file=sys.stderr)
        return 1
    pts = chart_series(aid)
    if pts is None:
        print(f"Could not fetch chart data for {aid}", file=sys.stderr)
        return 1
    if not pts:
        print(f"No chart data for app {aid} (unreleased / untracked).", file=sys.stderr)
        return 1

    if args.since:
        lo = int(datetime.strptime(args.since, "%Y-%m-%d")
                 .replace(tzinfo=timezone.utc).timestamp() * 1000)
        pts = [p for p in pts if p[0] >= lo]
    if args.until:
        hi = int(datetime.strptime(args.until, "%Y-%m-%d")
                 .replace(tzinfo=timezone.utc).timestamp() * 1000)
        pts = [p for p in pts if p[0] <= hi]

    if args.top:
        pts = sorted(pts, key=lambda p: p[1], reverse=True)[:args.top]
    else:
        pts = sorted(pts, key=lambda p: p[0])

    rows = [{"date": iso(ts), "players": v} for ts, v in pts]

    if args.json:
        print(json.dumps({"appid": aid, "series": rows}, indent=2))
    elif args.csv:
        w = csv.writer(sys.stdout)
        w.writerow(["date", "players"])
        for r in rows:
            w.writerow([r["date"], r["players"]])
    else:
        print(f"{'date (UTC)':<22} players")
        for r in rows:
            print(f"{r['date']:<22} {r['players']:,}")
    return 0


def cmd_dev(args):
    role = args.role
    found = {}
    roles = ["developer", "publisher"] if role == "any" else [role]
    for r in roles:
        for aid, title in store_by_role(r, args.name):
            found.setdefault(aid, title)

    if not found:
        print(f"No Steam apps found for {args.name!r} ({role}).", file=sys.stderr)
        return 1

    rows = []
    for aid in found:
        if args.peaks:
            info = summarize(aid, found[aid])
            rows.append(info)
            if not args.json:
                pk = info["all_time_peak"]
                extra = (f"{pk:,} @ {info['peak_at']}" if pk is not None
                         else (info["note"] or "n/a"))
                print(f"{aid:>9}  {(info['name'] or '?'):<32} {extra}")
        else:
            rows.append({"appid": aid, "name": found[aid]})

    if args.json:
        print(json.dumps(rows, indent=2))
    elif not args.peaks:
        print(f"{'appid':>9}  name")
        for r in rows:
            print(f"{r['appid']:>9}  {r['name']}")
    return 0


# ----------------------------------------------------------------------------
def main(argv=None):
    global _debug
    p = argparse.ArgumentParser(
        prog="steamcharts.py",
        description="Reliable Steam concurrent-player data (peak + current).",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__.split("USAGE")[1] if "USAGE" in __doc__ else None,
    )
    p.add_argument("--debug", action="store_true", help="print retry diagnostics to stderr")
    sub = p.add_subparsers(dest="cmd", required=True)

    def add_json(sp):
        sp.add_argument("--json", action="store_true", help="machine-readable output")

    sp = sub.add_parser("search", help="search Steam for apps by name")
    sp.add_argument("query")
    add_json(sp)
    sp.set_defaults(func=cmd_search)

    sp = sub.add_parser("resolve", help="name -> appid")
    sp.add_argument("query")
    add_json(sp)
    sp.set_defaults(func=cmd_resolve)

    sp = sub.add_parser("peak", help="all-time peak concurrents")
    sp.add_argument("appid", help="numeric appid or game name")
    add_json(sp)
    sp.set_defaults(func=cmd_peak)

    sp = sub.add_parser("app", help="full summary for one app")
    sp.add_argument("appid")
    add_json(sp)
    sp.set_defaults(func=cmd_app)

    sp = sub.add_parser("current", help="current concurrents (official Steam API)")
    sp.add_argument("appid")
    add_json(sp)
    sp.set_defaults(func=cmd_current)

    sp = sub.add_parser("series", help="peak-concurrency time series")
    sp.add_argument("appid")
    sp.add_argument("--csv", action="store_true", help="output CSV")
    sp.add_argument("--json", action="store_true", help="output JSON")
    sp.add_argument("--since", metavar="YYYY-MM-DD")
    sp.add_argument("--until", metavar="YYYY-MM-DD")
    sp.add_argument("--top", type=int, metavar="N", help="top N peaks instead of chronological")
    sp.set_defaults(func=cmd_series)

    sp = sub.add_parser("dev", help="all Steam apps by a developer/publisher")
    sp.add_argument("name")
    sp.add_argument("--role", choices=["developer", "publisher", "any"], default="any")
    sp.add_argument("--peaks", action="store_true", help="also fetch each game's all-time peak")
    add_json(sp)
    sp.set_defaults(func=cmd_dev)

    args = p.parse_args(argv)
    _debug = args.debug
    try:
        return args.func(args)
    except KeyboardInterrupt:
        return 130


if __name__ == "__main__":
    sys.exit(main())

Redaction report