refactor(stats): 统计一律读小时汇总，总览与图表口径统一（A2）

问题（实测）：overview/by_provider 读 usage_events（明细，只留 90 天），
timeline/model-timeline 读 usage_hourly（永久）。选「全部」时同屏两个数字
互相矛盾：200 天前的明细被清理后 overview=1、timeline=2。

- usage_hourly 补 reasoning_tokens / cached_tokens / cached_known 三列
  （SCHEMA_VERSION 5→6，走 _MIGRATION_COLUMNS；老库历史行补 0）
- overview / by_provider 改读小时汇总，与图表彻底同源
- record() 增量累加当前小时行：新请求立即可见，不必等 5 分钟一轮的 rollup
  （否则刚发生的请求统计页显示 0，比原来的口径矛盾更糟）；
  rollup_hourly 仍全量重算，两者结果一致（幂等，已加测试）
- 顺带修 C2：延迟均值分子分母配对——分子只累加成功请求
  （此前 SUM(latency_ms) 含失败、分母只有 ok_count，实测 200ms 被算成 650ms）
- 修 scripts/cleanup_invalid_stats.py 的 _DbAdapter：Database.transaction()
  重构后它少了该方法，--apply 直接 AttributeError（补该脚本的测试，此前零覆盖）

测试：总览与图表在 90 天外一致、reasoning/cached 聚合、均值只算成功、
老库补列保留历史行、增量与全量重算等价。

Author: RobbsLuo

TECHNICAL.md

@@ -78,8 +78,8 @@ coding2api/
 │   │   ├── retention.py         # 每 5 分钟：小时汇总重算（幂等）+ 90 天前明细清理
 │   │   └── runner.py            # 后台任务调度，接入应用生命周期
 │   ├── stats/
-│   │   ├── collector.py         # usage_events 写入（脱敏）
-│   │   └── query.py             # overview / by-provider / events 查询（events JOIN credentials 带凭证昵称）
+│   │   ├── collector.py         # usage_events 写入（脱敏）+ 小时汇总双写/重算
+│   │   └── query.py             # overview / by-provider / timeline / events 查询（前者读小时汇总，events JOIN credentials 带凭证昵称）
 │   └── api/
 │       ├── deps.py              # Services 容器 + require_api_key / session / csrf 依赖
 │       ├── chat.py              # POST /v1/chat/completions
@@ -286,7 +286,12 @@ class Scheduler:
 | 额度探测（quota_probe.py） | 启动立即跑一轮（不节流） + 每 `QUOTA_PROBE_MINUTES`（默认 60）分钟 | 探测上游剩余额度 → `credentials.quota_*` / `quota_expiry_ladder` / `health` 写回 |
 | token 预刷新（refresh.py） | 每 60 分钟 | 到期前 `REFRESH_SKEW_HOURS`（默认 24h）窗口内轮换 refresh token |
 | 每日签到（checkin.py） | 每 10 分钟（全天） | 成功即封账该凭证当日（`日期:scope`，进程内内存态，重启重建）；失败凭证持续重试，同账号多凭证共享一次 |
-| 明细清理（retention.py） | 每 5 分钟 | `usage_events` 全量重算小时汇总（幂等 upsert，最新小时滞后 ≤5 分钟）+ 90 天前明细清理 |
+| 明细清理（retention.py） | 每 5 分钟 | `usage_events` 全量重算小时汇总（幂等 upsert，与 record 的增量双写对账）+ 90 天前明细清理 |
+
+**清理切点必顶对齐到小时边界**（`purge_expired`）。这不是保守取值而是正确性要求：
+`rollup_hourly` 对整行是 REPLACE 语义，只有保证「仍有明细的小时保有全部明细」
+重算才精确；若把边界小时只删一半，下一轮 rollup 会把汇总行覆盖成剩下那一半，
+被删部分永久丢失（明细已不在）。代价：明细最多多留 1 小时。
 
 ---
 
@@ -360,6 +365,14 @@ fixture 断言两个方向：**解析正确**（样本 → 期望 Event）与**
 - **手写 SQL 而非 ORM**：8 张表规模下 ORM 收益为负
 - **polling OAuth 不转回调**（Q17=C）：上游协议决定；TRAE 回调走主端口 + PUBLIC_BASE_URL
 - **v1 无 Anthropic**（Q8=A）：Event 层已预留，v1.1 只加 `compat/anthropic/` 适配器
+- **统计一律以 `usage_hourly` 为准**：`overview` / `by_provider` / `timeline` /
+  `model-timeline` 均读小时汇总，只有 `events`（逐请求明细）读 `usage_events`。
+  统一口径是为了让选「全部」时总览与图表同值（明细只留 90 天，汇总永久）
+- **小时汇总双写**：`record()` 写入明细的同时增量累加当前小时行，所以新请求
+  立即可见于统计页（不依赖 5 分钟一轮的 rollup）；`rollup_hourly` 仍每 5 分钟
+  全量重算作对账，两者结果一致（幂等）
+- **延迟均值只算成功请求**：分子 `SUM(latency_ms WHERE ok=1)` 与分母 `ok_count`
+  配对；失败请求的耗时不能拉偏「典型耗时」（与图表口径一致）
 - **两套数据源共存（已知不一致）**：`overview` / `by_provider` 读 `usage_events`（即时，
   仅覆盖 90 天明细），`timeline` / `model-timeline` 读 `usage_hourly`（≤5 分钟滞后，永久）。
   时间范围 ≤90 天时两者一致（汇总由同一批明细算出）；选「全部」时总览会小于图表，

scripts/cleanup_invalid_stats.py

@@ -26,6 +26,7 @@
 import sqlite3
 import sys
 import time
+from contextlib import contextmanager
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
@@ -77,14 +78,24 @@ def sync_hourly(conn) -> None:
 
 
 class _DbAdapter:
-    """把裸连接包装成 Database.connect() 的形态（rollup 只用这一个方法）。"""
+    """把裸连接包装成 Database 的形态（rollup 只用 connect/transaction）。"""
 
     def __init__(self, conn) -> None:
         self._conn = conn
 
     def connect(self):
         return self._conn
 
+    @contextmanager
+    def transaction(self):
+        """与 Database.transaction 同语义：正常提交、异常回滚。"""
+        try:
+            yield self._conn
+        except BaseException:
+            self._conn.rollback()
+            raise
+        self._conn.commit()
+
 
 def backup(db_path: Path) -> Path:
     """删除前备份主库文件到同目录（.bak-<时间戳>），失败即中止。"""

src/db/migrate.py

@@ -16,7 +16,7 @@
 SCHEMA_NAME = "schema.sql"
 
 # 当前 schema 版本。新增列/表、删表时 +1，并在下方对应元组里补增量。
-SCHEMA_VERSION = 5
+SCHEMA_VERSION = 6
 
 # (表, 列定义)：历史库升级时逐条补列
 _MIGRATION_COLUMNS: tuple[tuple[str, str], ...] = (
@@ -26,6 +26,11 @@
      "ttfb_sum INTEGER NOT NULL DEFAULT 0"),  # 首字延迟聚合（图表维度，老库补 0）
     ("credentials",
      "quota_expiry_ladder TEXT"),  # 到期阶梯 JSON：选号按窗口内到期积分排序
+    # 总览改为读小时汇总后，这两项也必须能在小时表里聚合（老库历史行补 0，
+    # 历史小时的数值无法回填：明细已不在，只能接受旧时段显示 0）
+    ("usage_hourly", "reasoning_tokens INTEGER NOT NULL DEFAULT 0"),
+    ("usage_hourly", "cached_tokens INTEGER NOT NULL DEFAULT 0"),
+    ("usage_hourly", "cached_known INTEGER NOT NULL DEFAULT 0"),
 )
 
 # 已废弃的表：schema.sql 里已删定义，但老库里可能还留着，必须显式清理。

src/db/schema.sql

@@ -75,6 +75,9 @@ CREATE TABLE IF NOT EXISTS usage_hourly (
     ok_count      INTEGER NOT NULL DEFAULT 0,
     input_tokens  INTEGER NOT NULL DEFAULT 0,
     output_tokens INTEGER NOT NULL DEFAULT 0,
+    reasoning_tokens INTEGER NOT NULL DEFAULT 0,
+    cached_tokens INTEGER NOT NULL DEFAULT 0,      -- 命中缓存的输入 token 之和
+    cached_known  INTEGER NOT NULL DEFAULT 0,      -- 上报过 cached_tokens 的条数（=0 时该值不可信）
     credit_sum    REAL,
     credit_known  INTEGER NOT NULL DEFAULT 0,
     latency_sum   INTEGER NOT NULL DEFAULT 0,

src/stats/collector.py

@@ -109,6 +109,47 @@ def record(
                  event.output_tokens, event.reasoning_tokens, event.cached_tokens, event.credit,
                  event.latency_ms, event.ttfb_ms),
             )
+            # 当前小时增量累加：总览/图表都读小时表，不能等 5 分钟一轮的
+            # retention rollup 才可见（否则刚发生的请求统计页面显示 0）。
+            # 增量累加与全量重算等价：rollup 后续会把这一小时算成同样的值。
+            self._bump_hourly(conn, event)
+
+    @staticmethod
+    def _bump_hourly(conn, event: UsageEvent) -> None:
+        """把一条明细增量累加进所属小时行（同 key 则累加，不存在则插入）。"""
+        conn.execute(
+            """
+            INSERT INTO usage_hourly (hour_utc, username, provider, model, requests, ok_count,
+                                      input_tokens, output_tokens, reasoning_tokens,
+                                      cached_tokens, cached_known,
+                                      credit_sum, credit_known, latency_sum, ttfb_sum)
+            VALUES (?,?,?,?,1,?,?,?,?,?,?,?,?,?,?)
+            ON CONFLICT(hour_utc, username, provider, model) DO UPDATE SET
+                requests = requests + 1,
+                ok_count = ok_count + excluded.ok_count,
+                input_tokens = input_tokens + excluded.input_tokens,
+                output_tokens = output_tokens + excluded.output_tokens,
+                reasoning_tokens = reasoning_tokens + excluded.reasoning_tokens,
+                cached_tokens = cached_tokens + excluded.cached_tokens,
+                cached_known = cached_known + excluded.cached_known,
+                credit_sum = COALESCE(credit_sum, 0) + excluded.credit_sum,
+                credit_known = credit_known + excluded.credit_known,
+                latency_sum = latency_sum + excluded.latency_sum,
+                ttfb_sum = ttfb_sum + excluded.ttfb_sum
+            """,
+            # NULL 一律折成 0 参与累加：列定义是 NOT NULL DEFAULT 0，
+            # `x + NULL` 会交出 NULL，把整行污染掉
+            ((event.ts // 3600) * 3600, event.username, event.provider, event.model,
+             int(event.ok),
+             event.input_tokens or 0, event.output_tokens or 0,
+             event.reasoning_tokens or 0,
+             event.cached_tokens or 0,
+             0 if event.cached_tokens is None else 1,
+             event.credit or 0.0,
+             0 if event.credit is None else 1,
+             (event.latency_ms or 0) if event.ok else 0,
+             (event.ttfb_ms or 0) if event.ok else 0),
+        )
 
     def purge_expired(self, retention_days: int = 90, now: int | None = None) -> int:
         """明细保留 90 天；小时汇总永久（PROPOSAL §8）。
@@ -139,20 +180,28 @@ def rollup_hourly(self, since: int | None = None) -> int:
             cursor = conn.execute(
                 """
                 INSERT INTO usage_hourly (hour_utc, username, provider, model, requests, ok_count,
-                                          input_tokens, output_tokens, credit_sum, credit_known,
-                                          latency_sum, ttfb_sum)
+                                          input_tokens, output_tokens, reasoning_tokens,
+                                          cached_tokens, cached_known,
+                                          credit_sum, credit_known, latency_sum, ttfb_sum)
                 SELECT (ts / 3600) * 3600 AS hour_utc, username, provider, model,
                        COUNT(*), SUM(ok),
                        COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0),
-                       SUM(credit), SUM(CASE WHEN credit IS NULL THEN 0 ELSE 1 END),
-                       COALESCE(SUM(latency_ms), 0), COALESCE(SUM(ttfb_ms), 0)
+                       COALESCE(SUM(reasoning_tokens), 0),
+                       COALESCE(SUM(cached_tokens), 0),
+                       SUM(CASE WHEN cached_tokens IS NULL THEN 0 ELSE 1 END),
+                       COALESCE(SUM(credit), 0), SUM(CASE WHEN credit IS NULL THEN 0 ELSE 1 END),
+                       COALESCE(SUM(CASE WHEN ok = 1 THEN latency_ms END), 0),
+                       COALESCE(SUM(CASE WHEN ok = 1 THEN ttfb_ms END), 0)
                 FROM usage_events WHERE (? IS NULL OR ts >= ?)
                 GROUP BY hour_utc, username, provider, model
                 ON CONFLICT(hour_utc, username, provider, model) DO UPDATE SET
                     requests = excluded.requests,
                     ok_count = excluded.ok_count,
                     input_tokens = excluded.input_tokens,
                     output_tokens = excluded.output_tokens,
+                    reasoning_tokens = excluded.reasoning_tokens,
+                    cached_tokens = excluded.cached_tokens,
+                    cached_known = excluded.cached_known,
                     credit_sum = excluded.credit_sum,
                     credit_known = excluded.credit_known,
                     latency_sum = excluded.latency_sum,

src/stats/query.py

@@ -52,48 +52,61 @@ def _where(*, username: str | None = None, provider: str | None = None,
 
     def overview(self, *, username: str | None = None, provider: str | None = None,
                  since: int | None = None) -> dict[str, Any]:
-        """总览。admin 传 username=None 看全局；普通用户只看自己。"""
-        where, params = self._where(username=username, provider=provider, since=since)
+        """总览。admin 传 username=None 看全局；普通用户只看自己。
+
+        数据源是 `usage_hourly`（与 timeline/model-timeline 同源）：
+        明细只留 90 天，读明细会让选「全部」时总览小于图表；小时汇总永久保留。
+        代价：最近 ≤5 分钟未进汇总的请求不计入（retention 每 5 分钟 rollup），
+        刷新一次即可。延迟类均值统一按 `ok_count` 归一（与图表同口径）。
+        """
+        where, params = self._where(username=username, provider=provider,
+                                    since=since, since_col="hour_utc")
 
         row = self._db.connect().execute(
             f"""
-            SELECT COUNT(*) AS requests,
-                   COALESCE(SUM(ok), 0) AS ok_count,
+            SELECT COALESCE(SUM(requests), 0) AS requests,
+                   COALESCE(SUM(ok_count), 0) AS ok_count,
                    COALESCE(SUM(input_tokens), 0) AS input_tokens,
                    COALESCE(SUM(output_tokens), 0) AS output_tokens,
                    COALESCE(SUM(reasoning_tokens), 0) AS reasoning_tokens,
                    SUM(cached_tokens) AS cached_tokens,
-                   SUM(CASE WHEN cached_tokens IS NULL THEN 0 ELSE 1 END) AS cached_known,
-                   SUM(credit) AS credit_sum,
-                   SUM(CASE WHEN credit IS NULL THEN 0 ELSE 1 END) AS credit_known,
-                   COALESCE(AVG(latency_ms), 0) AS avg_latency,
-                   COALESCE(AVG(ttfb_ms), 0) AS avg_ttfb
-            FROM usage_events {where}
+                   SUM(cached_known) AS cached_known,
+                   SUM(credit_sum) AS credit_sum,
+                   SUM(credit_known) AS credit_known,
+                   SUM(latency_sum) AS latency_sum,
+                   SUM(ttfb_sum) AS ttfb_sum
+            FROM usage_hourly {where}
             """, params).fetchone()
         requests = row["requests"] or 0
+        ok_count = row["ok_count"] or 0
+        # 均值口径：只除成功请求（失败请求的 latency 会拉偏"典型耗时"，
+        # 且与图表 latency_sum/ok_count 同一致），无成功则 None
         return {
             "requests": requests,
-            "ok_count": row["ok_count"],
-            "success_rate": (row["ok_count"] / requests) if requests else None,
+            "ok_count": ok_count,
+            "success_rate": (ok_count / requests) if requests else None,
             "input_tokens": row["input_tokens"],
             "output_tokens": row["output_tokens"],
             "reasoning_tokens": row["reasoning_tokens"],
             # 缓存命中：任一明细上报过才可信，否则 None（前端显示 —）
             "cached_tokens": row["cached_tokens"] if row["cached_known"] else None,
             "credit": row["credit_sum"] if row["credit_known"] else None,
-            "avg_latency_ms": round(row["avg_latency"]) if requests else None,
-            "avg_ttfb_ms": round(row["avg_ttfb"]) if requests else None,
+            "avg_latency_ms": round(row["latency_sum"] / ok_count) if ok_count else None,
+            "avg_ttfb_ms": round(row["ttfb_sum"] / ok_count) if ok_count else None,
         }
 
     def by_provider(self, *, username: str | None = None,
                     since: int | None = None) -> list[dict[str, Any]]:
-        where, params = self._where(username=username, since=since)
+        """按渠道聚合（同样读小时汇总，与总览/图表同源）。"""
+        where, params = self._where(username=username, since=since, since_col="hour_utc")
         rows = self._db.connect().execute(
             f"""
-            SELECT provider, COUNT(*) AS requests, COALESCE(SUM(ok), 0) AS ok_count,
+            SELECT provider, COALESCE(SUM(requests), 0) AS requests,
+                   COALESCE(SUM(ok_count), 0) AS ok_count,
                    COALESCE(SUM(input_tokens), 0) AS input_tokens,
-                   COALESCE(SUM(output_tokens), 0) AS output_tokens, SUM(credit) AS credit_sum
-            FROM usage_events {where} GROUP BY provider ORDER BY provider
+                   COALESCE(SUM(output_tokens), 0) AS output_tokens,
+                   SUM(credit_sum) AS credit_sum
+            FROM usage_hourly {where} GROUP BY provider ORDER BY provider
             """, params).fetchall()
         return [
             {"provider": row["provider"], "requests": row["requests"],