metrics: energy and carbon per day, from measured GreenPT factors

Implements the approach Nathan settled on in co-op/support#26: per-token
estimates based on what GreenPT publishes, rather than waiting on the upstream
LiteLLM `impact` passthrough.

- New daily columns: out_tokens, energy_wh, co2e_g.
- Tokens come from OUR OWN spend logs (completion_tokens per model per day), so
  member-visible impact does not depend on any client persisting an `impact`
  object. GreenPT's metric counts OUTPUT tokens only — that is the unit — so
  input tokens are excluded deliberately.
- Factors live in one table with a source per row. The greenpt/* rows are
  MEASURED: fitted from live calls to their API across prompt sizes
  (energy = overhead + marginal x output tokens, R2 >= 0.999; carbon from the
  grid intensity their own emissions imply — ug/mWh and g/kWh are the same
  ratio). Everything else falls back to a documented order-of-magnitude proxy,
  so a day's total is never silently missing models.
- Migration: CREATE IF NOT EXISTS never adds columns, so both the script and
  the app ALTER TABLE on connect.

Refining a factor is a data edit in IMPACT_FACTORS, not a code change.

app/main.py

@@ -622,9 +622,19 @@ def get_db() -> sqlite3.Connection:
         "matrix_space_members INTEGER, "
         "spend_usd REAL, "
         "requests INTEGER, "
+        "out_tokens INTEGER, "
+        "energy_wh REAL, "
+        "co2e_g REAL, "
         "recorded_at TEXT"
         ")"
     )
+    # CREATE IF NOT EXISTS never adds columns to an existing table, so migrate:
+    # impact columns arrived after the table did (see co-op/support#26).
+    for _col, _decl in (("out_tokens", "INTEGER"), ("energy_wh", "REAL"), ("co2e_g", "REAL")):
+        try:
+            conn.execute(f"ALTER TABLE metrics_daily ADD COLUMN {_col} {_decl}")
+        except sqlite3.OperationalError:
+            pass  # already present
     # Member-managed API keys (created via the broker). Each row is one API key
     # under the member's team, tracked so we can list/revoke by name without
     # exposing the chat key. `sk_token` is the full sk- key (revealed once).
@@ -3059,15 +3069,19 @@ def metrics_series(days: int = 90) -> list[dict]:
 
     One query, two callers: the admin panel (via /admin/metrics/series) and the
     member dashboard (via /broker/metrics). Columns are cumulative as of that
-    day except spend_usd/requests, which are that day only; NULL means the
-    source could not be read that day.
+    day except spend_usd/requests/out_tokens/energy_wh/co2e_g, which are that
+    day only; NULL means the source could not be read that day.
+
+    energy_wh and co2e_g are computed in metrics_snapshot.py from the day's
+    output tokens against a factor table: measured for GreenPT models (their API
+    returns an `impact` object), estimated for the rest. See co-op/support#26.
     """
     limit = max(1, min(int(days or 90), 730))
     conn = get_db()
     try:
         rows = conn.execute(
             "SELECT day, members_active, members_total, members_new, newsletter_subs, "
-            "matrix_space_members, spend_usd, requests, recorded_at "
+            "matrix_space_members, spend_usd, requests, out_tokens, energy_wh, co2e_g, recorded_at "
             "FROM metrics_daily ORDER BY day DESC LIMIT ?",
             (limit,),
         ).fetchall()
@@ -3082,6 +3096,9 @@ def metrics_series(days: int = 90) -> list[dict]:
         "matrix_space_members",
         "spend_usd",
         "requests",
+        "out_tokens",
+        "energy_wh",
+        "co2e_g",
         "recorded_at",
     ]
     return [dict(zip(cols, row)) for row in reversed(rows)]

metrics_snapshot.py

@@ -57,8 +57,17 @@ def connect() -> sqlite3.Connection:
                matrix_space_members INTEGER,
                spend_usd REAL,
                requests INTEGER,
+               out_tokens INTEGER,
+               energy_wh REAL,
+               co2e_g REAL,
                recorded_at TEXT)"""
     )
+    # CREATE IF NOT EXISTS never adds columns to an existing table, so migrate.
+    for col, decl in (("out_tokens", "INTEGER"), ("energy_wh", "REAL"), ("co2e_g", "REAL")):
+        try:
+            conn.execute(f"ALTER TABLE metrics_daily ADD COLUMN {col} {decl}")
+        except sqlite3.OperationalError:
+            pass  # already present
     return conn
 
 
@@ -68,6 +77,34 @@ def _get_json(url: str, headers: dict | None = None):
         return json.loads(resp.read() or b"{}")
 
 
+# ---------------------------------------------------------------- impact factors
+# Energy per 1,000 OUTPUT tokens, and the carbon intensity of the electricity.
+#
+# The greenpt/* rows are MEASURED. GreenPT returns an `impact` object on every
+# call (energy in watt-milliseconds, emissions in µgCO2e), so these were fitted
+# from live calls to their API: energy regressed on output tokens across prompt
+# sizes (R² >= 0.999), carbon taken as the grid intensity their own emissions
+# imply (µg/mWh and g/kWh are the same ratio). Their published unit is mWh per
+# 100 output tokens, and only OUTPUT tokens count — hence completion_tokens.
+#
+# Everything else is an ESTIMATE and is labelled as one. Providers that do not
+# disclose impact get IMPACT_PROXY: an order-of-magnitude figure near the middle
+# of the measured range with an EU grid average. It exists so a day's total is
+# never silently missing models, not because we know the number. Refining these
+# is a data edit here, not a code change — see co-op/support#26.
+MEASURED = "measured (GreenPT impact API, fitted 2026-10-28)"
+SAME_FAMILY = "measured for glm-5.3-flash; assumed equal for glm-5.3"
+ESTIMATED = "estimated (order-of-magnitude proxy; provider does not disclose)"
+
+# Longest prefix must come first: matching is startswith().
+IMPACT_FACTORS = {
+    "greenpt/deepseek-v4.1-flash": {"wh_per_1k_out": 0.3641, "gco2e_per_kwh": 193.0, "source": MEASURED},
+    "greenpt/glm-5.3-flash":       {"wh_per_1k_out": 1.0782, "gco2e_per_kwh": 193.0, "source": MEASURED},
+    "greenpt/glm-5.3":             {"wh_per_1k_out": 1.0782, "gco2e_per_kwh": 193.0, "source": SAME_FAMILY},
+    "greenpt/green-r":             {"wh_per_1k_out": 0.5535, "gco2e_per_kwh": 23.6,  "source": MEASURED},
+}
+IMPACT_PROXY = {"wh_per_1k_out": 0.5000, "gco2e_per_kwh": 250.0, "source": ESTIMATED}
+
 # ---------------------------------------------------------------- sources
 
 def listmonk_joins() -> list[str] | None:
@@ -226,6 +263,78 @@ def spend_by_day(since: dt.date | None = None) -> dict[str, tuple[float, int]]:
     return out
 
 
+def tokens_by_day_model(since: dt.date | None = None) -> dict[str, dict[str, int]]:
+    """{day: {model: output_tokens}} from LiteLLM's spend logs.
+
+    OUTPUT tokens only — that is the unit GreenPT's published energy metric is
+    normalised to ("tokens = computational distance"; input tokens are excluded
+    so comparisons hold across prompts and models). Read from our own logs, so
+    member-visible impact does not depend on any client persisting anything.
+    """
+    try:
+        import psycopg2
+    except Exception as exc:  # noqa: BLE001
+        print(f"metrics: psycopg2 unavailable ({exc})")
+        return {}
+    try:
+        conn = psycopg2.connect(
+            host=os.environ.get("LITELLM_DB_HOST"),
+            port=os.environ.get("LITELLM_DB_PORT"),
+            user=os.environ.get("LITELLM_DB_USER"),
+            password=os.environ.get("LITELLM_DB_PASSWORD"),
+            dbname=os.environ.get("LITELLM_DB_NAME"),
+            connect_timeout=10,
+            sslmode="prefer",
+        )
+    except Exception as exc:  # noqa: BLE001
+        print(f"metrics: LiteLLM DB connect failed ({exc}) — tokens left NULL")
+        return {}
+    try:
+        where, params = "", []
+        if since:
+            where, params = 'WHERE "startTime" >= %s', [since.isoformat()]
+        with conn.cursor() as cur:
+            cur.execute(
+                f"""SELECT to_char("startTime", 'YYYY-MM-DD') AS day, model,
+                           COALESCE(SUM(completion_tokens), 0)
+                    FROM "LiteLLM_SpendLogs" {where}
+                    GROUP BY 1, 2""",
+                params,
+            )
+            out: dict[str, dict[str, int]] = {}
+            for day, model, toks in cur.fetchall():
+                out.setdefault(day, {})[model or "unknown"] = int(toks)
+    finally:
+        conn.close()
+    return out
+
+
+def impact_for(models: dict[str, int]) -> tuple[float | None, float | None]:
+    """(Wh, gCO2e) for a day's {model: output_tokens}.
+
+    Factors live in IMPACT_FACTORS. Unknown models fall back to the documented
+    proxy row rather than being skipped, so a total is never quietly missing a
+    model — but the proxy is labelled as an order-of-magnitude estimate and the
+    docs say so. Energy: Wh per 1,000 output tokens. Carbon: gCO2e per kWh.
+    """
+    if not models:
+        return None, None
+    wh = 0.0
+    co2 = 0.0
+    for model, toks in models.items():
+        if not toks:
+            continue
+        f = None
+        for prefix, factor in IMPACT_FACTORS.items():
+            if model == prefix or model.startswith(prefix):
+                f = factor
+                break
+        f = f or IMPACT_PROXY
+        wh += toks / 1000.0 * f["wh_per_1k_out"]
+        co2 += toks / 1000.0 * f["wh_per_1k_out"] / 1e6 * f["gco2e_per_kwh"]
+    return round(wh, 6), round(co2, 6)
+
+
 # ---------------------------------------------------------------- writes
 
 def upsert(day: str, row: dict) -> None:
@@ -234,8 +343,9 @@ def upsert(day: str, row: dict) -> None:
         conn.execute(
             """INSERT INTO metrics_daily
                    (day, members_active, members_total, members_new, newsletter_subs,
-                    matrix_space_members, spend_usd, requests, recorded_at)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    matrix_space_members, spend_usd, requests, out_tokens, energy_wh,
+                    co2e_g, recorded_at)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(day) DO UPDATE SET
                    members_active=excluded.members_active,
                    members_total=excluded.members_total,
@@ -244,6 +354,9 @@ def upsert(day: str, row: dict) -> None:
                    matrix_space_members=excluded.matrix_space_members,
                    spend_usd=excluded.spend_usd,
                    requests=excluded.requests,
+                   out_tokens=excluded.out_tokens,
+                   energy_wh=excluded.energy_wh,
+                   co2e_g=excluded.co2e_g,
                    recorded_at=excluded.recorded_at""",
             (
                 day,
@@ -254,6 +367,9 @@ def upsert(day: str, row: dict) -> None:
                 row.get("matrix_space_members"),
                 row.get("spend_usd"),
                 row.get("requests"),
+                row.get("out_tokens"),
+                row.get("energy_wh"),
+                row.get("co2e_g"),
                 dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
             ),
         )
@@ -268,8 +384,13 @@ def snapshot(day: str | None = None) -> dict:
     subs = from_listmonk()
     joins = listmonk_joins()
     space = from_matrix()
-    spend = spend_by_day(since=dt.date.fromisoformat(day) - dt.timedelta(days=1))
+    since = dt.date.fromisoformat(day) - dt.timedelta(days=1)
+    spend = spend_by_day(since=since)
     today_spend = spend.get(day, (None, None))
+    day_models = (tokens_by_day_model(since=since).get(day)) or {}
+    out_tokens, energy_wh, co2e_g = (
+        (sum(day_models.values()), *impact_for(day_models)) if day_models else (None, None, None)
+    )
     row = {
         "members_active": m["active"],
         "members_total": m["total"],
@@ -278,6 +399,9 @@ def snapshot(day: str | None = None) -> dict:
         "matrix_space_members": space,
         "spend_usd": today_spend[0],
         "requests": today_spend[1],
+        "out_tokens": out_tokens,
+        "energy_wh": energy_wh,
+        "co2e_g": co2e_g,
     }
     upsert(day, row)
     print(f"metrics: recorded {day} -> {json.dumps(row)}")
@@ -296,6 +420,7 @@ def backfill() -> None:
     m = from_sqlite()
     joins = sorted(listmonk_joins() or [])
     spend = spend_by_day()
+    tokens = tokens_by_day_model()
     if not joins:
         print("metrics: no membership history (listmonk unavailable) — past days left NULL for members")
     if not joins and not spend:
@@ -307,7 +432,7 @@ def backfill() -> None:
     last = today()
     print(
         f"metrics: backfilling {first}..{last} "
-        f"({len(joins)} join dates, {len(spend)} spend days)"
+        f"({len(joins)} join dates, {len(spend)} spend days, {len(tokens)} token days)"
     )
     day = dt.date.fromisoformat(first)
     end = dt.date.fromisoformat(last)
@@ -323,6 +448,10 @@ def backfill() -> None:
             "spend_usd": spend.get(key, (None, None))[0],
             "requests": spend.get(key, (None, None))[1],
         }
+        day_models = tokens.get(key) or {}
+        if day_models:
+            row["out_tokens"] = sum(day_models.values())
+            row["energy_wh"], row["co2e_g"] = impact_for(day_models)
         if key == today():
             # today's row carries the exact live metrics
             row.update(