climperor/web/fetch_item_counter_stats.py

"""Fetch adjusted enemy-item evidence for each hero from OpenDota Explorer.

For every hero, this compares enemy-team final-inventory item presence against
the same item's baseline across all teams in the same recent match window:

  buy_lift  = P(item | against hero) - P(item | any team)
  win_delta = P(win | item, against hero) - P(win | item, any team)

These are observational signals, not causal counter claims. The output is an
optional cache consumed by item_fears.py to corroborate and gently rerank its
mechanism-based candidates.

Usage:
  python fetch_item_counter_stats.py
  python fetch_item_counter_stats.py --matches 20000 --min-games 100
  python fetch_item_counter_stats.py --print-sql
  python fetch_item_counter_stats.py --soft-fail
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import argparse
import json
import os
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

from shared.grid import hero_table
from shared.http_utils import write_json_atomic
from shared.paths import DATA

API = "https://api.opendota.com/api/explorer"
ITEMS_META = DATA / "items_meta.json"
OUT = DATA / "item_counter_stats.json"


def build_sql(matches: int, min_games: int) -> str:
    """Build one bounded aggregate query; integer args are caller-validated."""
    return f"""
WITH recent_ids AS (
  SELECT DISTINCT match_id
  FROM player_matches
  ORDER BY match_id DESC
  LIMIT {matches}
),
recent AS (
  SELECT m.match_id, m.radiant_win
  FROM matches m
  JOIN recent_ids r ON r.match_id = m.match_id
),
players AS (
  SELECT
    p.match_id,
    p.hero_id,
    (p.player_slot < 128) AS is_radiant,
    p.item_0, p.item_1, p.item_2,
    p.item_3, p.item_4, p.item_5
  FROM player_matches p
  JOIN recent r ON r.match_id = p.match_id
  WHERE p.hero_id IS NOT NULL
),
teams AS (
  SELECT DISTINCT match_id, is_radiant
  FROM players
),
team_outcomes AS (
  SELECT
    t.match_id,
    t.is_radiant,
    CASE WHEN t.is_radiant THEN r.radiant_win ELSE NOT r.radiant_win END AS won
  FROM teams t
  JOIN recent r ON r.match_id = t.match_id
),
team_items AS (
  SELECT DISTINCT
    p.match_id,
    p.is_radiant,
    x.item_id
  FROM players p
  CROSS JOIN LATERAL (
    VALUES (p.item_0), (p.item_1), (p.item_2),
           (p.item_3), (p.item_4), (p.item_5)
  ) AS x(item_id)
  WHERE x.item_id IS NOT NULL AND x.item_id > 0
),
hero_totals AS (
  SELECT hero_id, COUNT(*) AS target_games
  FROM players
  GROUP BY hero_id
),
global_total AS (
  SELECT COUNT(*) AS team_games
  FROM team_outcomes
),
global_items AS (
  SELECT
    ti.item_id,
    COUNT(*) AS item_games,
    SUM(CASE WHEN o.won THEN 1 ELSE 0 END) AS item_wins
  FROM team_items ti
  JOIN team_outcomes o
    ON o.match_id = ti.match_id
   AND o.is_radiant = ti.is_radiant
  GROUP BY ti.item_id
),
target_items AS (
  SELECT
    p.hero_id,
    ti.item_id,
    COUNT(*) AS item_games,
    SUM(CASE WHEN o.won THEN 1 ELSE 0 END) AS item_wins
  FROM players p
  JOIN team_items ti
    ON ti.match_id = p.match_id
   AND ti.is_radiant <> p.is_radiant
  JOIN team_outcomes o
    ON o.match_id = ti.match_id
   AND o.is_radiant = ti.is_radiant
  GROUP BY p.hero_id, ti.item_id
  HAVING COUNT(*) >= {min_games}
)
SELECT
  t.hero_id,
  t.item_id,
  h.target_games,
  t.item_games,
  t.item_wins,
  g.team_games AS global_team_games,
  gi.item_games AS global_item_games,
  gi.item_wins AS global_item_wins
FROM target_items t
JOIN hero_totals h ON h.hero_id = t.hero_id
JOIN global_items gi ON gi.item_id = t.item_id
CROSS JOIN global_total g
ORDER BY t.hero_id, t.item_games DESC
""".strip()


def explorer(sql: str, *, timeout: int = 240) -> dict:
    params = {"sql": sql}
    api_key = os.environ.get("OPENDOTA_API_KEY", "").strip()
    if api_key:
        params["api_key"] = api_key
    url = API + "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"User-Agent": "climperor"})
    with urllib.request.urlopen(req, timeout=timeout) as response:
        payload = json.loads(response.read().decode())
    if not isinstance(payload, dict):
        raise RuntimeError("OpenDota Explorer returned a non-object payload")
    if payload.get("error") or payload.get("err"):
        raise RuntimeError(str(payload.get("error") or payload.get("err")))
    return payload


def item_catalog() -> dict[int, dict]:
    raw = json.loads(ITEMS_META.read_text(encoding="utf-8"))
    out: dict[int, dict] = {}
    for raw_id, row in (raw.get("items") or {}).items():
        if not isinstance(row, dict) or not row.get("key"):
            continue
        item_id = int(row.get("id") or raw_id)
        out[item_id] = {
            "key": str(row["key"]),
            "name_loc": str(row.get("name_loc") or row.get("dname") or row["key"]),
            "cost": int(row.get("cost") or 0),
        }
    return out


def build_payload(
    rows: list[dict],
    *,
    matches: int,
    min_games: int,
    min_cost: int,
) -> dict:
    heroes = {int(row["id"]): row["key"] for row in hero_table()}
    items = item_catalog()
    by_hero: dict[str, list[dict]] = {key: [] for key in heroes.values()}

    for row in rows:
        hero_key = heroes.get(int(row["hero_id"]))
        item = items.get(int(row["item_id"]))
        if not hero_key or not item or item["cost"] < min_cost:
            continue

        target_games = int(row["target_games"])
        item_games = int(row["item_games"])
        item_wins = int(row["item_wins"])
        global_team_games = int(row["global_team_games"])
        global_item_games = int(row["global_item_games"])
        global_item_wins = int(row["global_item_wins"])
        if not all((target_games, item_games, global_team_games, global_item_games)):
            continue

        buy_rate = item_games / target_games
        global_buy_rate = global_item_games / global_team_games
        win_rate = item_wins / item_games
        global_win_rate = global_item_wins / global_item_games
        by_hero[hero_key].append(
            {
                "item": item["key"],
                "name_loc": item["name_loc"],
                "games": item_games,
                "wins": item_wins,
                "target_games": target_games,
                "global_team_games": global_team_games,
                "global_item_games": global_item_games,
                "global_item_wins": global_item_wins,
                "buy_rate": round(buy_rate, 6),
                "global_buy_rate": round(global_buy_rate, 6),
                "buy_lift": round(buy_rate - global_buy_rate, 6),
                "win_rate": round(win_rate, 6),
                "global_win_rate": round(global_win_rate, 6),
                "win_delta": round(win_rate - global_win_rate, 6),
            }
        )

    for entries in by_hero.values():
        entries.sort(
            key=lambda row: (
                -float(row["buy_lift"]),
                -float(row["win_delta"]),
                -int(row["games"]),
                str(row["item"]),
            )
        )

    return {
        "meta": {
            "source": "opendota_explorer",
            "attribution": "https://www.opendota.com",
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "window_matches": matches,
            "min_games": min_games,
            "min_cost": min_cost,
            "method": "enemy final inventory vs same-item global team baseline",
            "caveat": "observational; adjusted for item baseline, not duration, rank, role, or economy",
        },
        "by_hero": by_hero,
    }


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--matches", type=int, default=20_000)
    ap.add_argument("--min-games", type=int, default=100)
    ap.add_argument("--min-cost", type=int, default=1_400)
    ap.add_argument("--out", type=Path, default=OUT)
    ap.add_argument("--print-sql", action="store_true")
    ap.add_argument(
        "--soft-fail",
        action="store_true",
        help="keep an existing cache and exit 0 when OpenDota is unavailable",
    )
    args = ap.parse_args()

    if args.matches < 1 or args.min_games < 1 or args.min_cost < 0:
        raise SystemExit("matches/min-games must be positive and min-cost non-negative")
    sql = build_sql(args.matches, args.min_games)
    if args.print_sql:
        print(sql)
        return
    if not ITEMS_META.is_file():
        raise SystemExit(f"missing {ITEMS_META}; run: python fetch_items_meta.py")

    try:
        payload = explorer(sql)
        rows = payload.get("rows") or []
        if not isinstance(rows, list) or not rows:
            raise RuntimeError("OpenDota Explorer returned no rows")
        output = build_payload(
            rows,
            matches=args.matches,
            min_games=args.min_games,
            min_cost=args.min_cost,
        )
        write_json_atomic(args.out, output)
        nonempty = sum(bool(v) for v in output["by_hero"].values())
        print(
            f"done: {len(rows)} rows, {nonempty} heroes -> {args.out}",
            flush=True,
        )
    except (OSError, TimeoutError, urllib.error.URLError, RuntimeError) as exc:
        if args.soft_fail:
            state = "keeping existing cache" if args.out.is_file() else "no cache available"
            print(f"warn: item counter stats unavailable ({exc}); {state}", flush=True)
            return
        raise


if __name__ == "__main__":
    main()