diff --git a/app/main.py b/app/main.py index 1febf8f..d6f9b7b 100644 --- a/app/main.py +++ b/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)] diff --git a/metrics_snapshot.py b/metrics_snapshot.py index f2ea1a8..20122d7 100644 --- a/metrics_snapshot.py +++ b/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(