feat(ad-revenue): 收益报表分页/场景筛选/倒序,并修复信息流金币审计复算口径

收益报表(/admin/api/ad-revenue-report):
- 明细改按时间倒序;新增 offset 真分页(limit 作每页大小、total 为全量),可翻页看当前筛选下全部数据,突破原 1000 条上限。
- 「场景」(feed_scene)下推后端做全局筛选,同时作用于明细/合计/daily·hourly 趋势(原为前端仅过滤明细)。
- 新增全量 hourly 序列,按小时趋势改用它,不再受分页截断影响。

修复信息流金币审计复算口径漂移(ad_audit):
- 发奖侧 grant_feed_reward 早已是「一条广告=1份、LT 按账号累计条数(COUNT)」,但审计仍按 unit_count 逐份累加 + SUM(unit_count) 做 LT 基线,导致单条停留>20s(份数>1)时应发虚高、必然「✗ 不符」。
- 审计改为每条 granted 按 1 份复算、LT 基线用 COUNT,与发奖对齐;金币审计页与收益报表(复用同一复算)一并恢复正确。纯复算口径修正,不改实际发奖、不动钱。

文档同步更新 admin-ad-revenue-report.md。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/admin/repositories/ad_audit.py

@@ -4,9 +4,9 @@
 - 看视频:每条 granted = 1 份,第 N 份 = 该用户 granted 的 reward_video **账号累计**顺序号
   (与 ad_reward.grant_ad_reward 里 `_granted_cumulative + 1` 一致;LT 因子不按天重置,
   故复算时要把当日序号叠加上该用户在本日**之前**的累计已发份数)。
-- 信息流:每条按 unit_count 份逐份累加,LT 序号 = 该用户 granted 份数**账号累计**
-  (与 ad_feed_reward._unit_reward_total 的 existing_units 一致;同样不按天重置,
-  复算需叠加本日之前的累计份数)。
+- 信息流:**每条 granted = 1 份**(与 ad_feed_reward.grant_feed_reward 同口径:看满一份即发该条
+  满额,**不按 unit_count 逐份累加**),LT 序号 = 该用户 granted **条数**账号累计
+  (与 ad_feed_reward.granted_unit_total 的 COUNT 一致;不按天重置,复算需叠加本日之前的累计条数)。
 
 非 granted(capped/ecpm_missing)不占用份序号、应发恒 0,据此校验闸口是否确实没发。
 """
@@ -108,14 +108,18 @@ def _reward_video_rows(
     return rows
 
 
-def _feed_prior_granted_units(
+def _feed_prior_granted_count(
     db: Session, *, date: str, user_id: int | None
 ) -> dict[int, int]:
-    """各用户在 date **之前** granted 的信息流份数累计,作为当日复算的 LT 序号起点。"""
+    """各用户在 date **之前** granted 的信息流**条数**累计,作为当日复算的 LT 序号起点。
+
+    与发奖侧 ad_feed_reward.granted_unit_total(COUNT status=granted)对齐:一条广告 = 1 份,
+    LT 按账号累计**条数**递进。**不再用 SUM(unit_count)**——那是「一条按时长折多份」的过时口径,
+    与现行发奖(每条 1 份)漂移,会让 unit_count>1 的记录复算虚高、对账恒「不符」。"""
     stmt = (
         select(
             AdFeedRewardRecord.user_id,
-            func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0),
+            func.count(),
         )
         .where(
             AdFeedRewardRecord.reward_date < date,
@@ -144,10 +148,11 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
 def _feed_rows(
     db: Session, *, date: str, user_id: int | None, scene: str | None = None
 ) -> list[dict]:
-    """信息流记录复算。granted 记录逐份累加,LT 序号沿用账号累计份数(含本日之前)。
+    """信息流记录复算。**每条 granted = 1 份**(与发奖同口径,不按 unit_count 累加),
+    LT 序号沿用账号累计**条数**(含本日之前)。
 
-    **关键:LT 因子账号累计按全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
-    故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_units 累加;scene 只决定
+    **关键:LT 因子账号累计按全表 granted 条数累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
+    故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_count 累加;scene 只决定
     哪些行被**留下展示**(由 _feed_scene_matches 判断),不影响累计基线,保证复算序号与正式发奖一致。
     """
     stmt = (
@@ -158,23 +163,20 @@ def _feed_rows(
     if user_id is not None:
         stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
 
-    # 本日之前的累计份数做起点,与 _unit_reward_total 的 existing_units(累计)对齐
-    granted_units: dict[int, int] = _feed_prior_granted_units(db, date=date, user_id=user_id)
+    # 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
+    granted_count: dict[int, int] = _feed_prior_granted_count(db, date=date, user_id=user_id)
     rows: list[dict] = []
     for rec in db.execute(stmt).scalars():
         keep = _feed_scene_matches(rec, scene)  # 累计照常推进,这里只决定是否展示本行
         if rec.status == "granted":
-            existing = granted_units.get(rec.user_id, 0)
-            units = rec.unit_count
-            granted_units[rec.user_id] = existing + units
+            # 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
+            # nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
+            # (即便 scene 不匹配不展示也要 +1,保证序号与正式发奖一致)。
+            nth = granted_count.get(rec.user_id, 0) + 1
+            granted_count[rec.user_id] = nth
             if not keep:
                 continue
-            expected = sum(
-                rewards.calculate_ad_reward_coin(rec.ecpm_raw, existing + offset)
-                for offset in range(1, units + 1)
-            )
-            start = existing + 1 if units > 0 else None
-            end = existing + units if units > 0 else None
+            expected = rewards.calculate_ad_reward_coin(rec.ecpm_raw, nth)
             rows.append({
                 "scene": "feed",
                 "ad_type": rec.ad_type or "feed",
@@ -188,11 +190,11 @@ def _feed_rows(
                 "status": rec.status,
                 "ecpm": rec.ecpm_raw,
                 "ecpm_factor": rewards.ad_ecpm_factor(rewards.parse_ecpm_yuan(rec.ecpm_raw)),
-                "units": units,
-                "lt_index_start": start,
-                "lt_index_end": end,
-                "lt_factor_start": rewards.ad_lt_factor(start) if start else None,
-                "lt_factor_end": rewards.ad_lt_factor(end) if end else None,
+                "units": 1,
+                "lt_index_start": nth,
+                "lt_index_end": nth,
+                "lt_factor_start": rewards.ad_lt_factor(nth),
+                "lt_factor_end": rewards.ad_lt_factor(nth),
                 "expected_coin": expected,
                 "actual_coin": rec.coin,
                 "matched": expected == rec.coin,

app/admin/repositories/ad_revenue.py

@@ -84,14 +84,19 @@ def ad_revenue_report(
     date_to: str,
     user_id: int | None = None,
     ad_type: str | None = None,
+    feed_scene: str | None = None,
     granularity: str = "day",
     limit: int = 500,
+    offset: int = 0,
 ) -> dict:
     """日期区间(北京时间,闭区间)**逐条广告事件**列表 + 发奖对账。单日时 date_from==date_to。
 
     每个 item = 一次广告事件(展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行)。
-    ad_type: None=全部 / reward_video / feed / draw。granularity=hour 时每行带北京小时(由各自时间算)。
-    limit 只截断 items(事件明细),total 与 total_* / daily 在全量上统计,数字始终可信。
+    ad_type: None=全部 / reward_video / feed / draw。feed_scene: None=全部 /
+    comparison / coupon / welfare,作为全局筛选(同时作用于明细、合计与 daily/hourly 趋势)。
+    granularity=hour 时每行带北京小时(由各自时间算),并额外返回全量 hourly 序列。
+    事件按时间倒序(新→旧)排列;limit/offset 对排序后的全量做分页切片(items 为当前页),
+    total 与 total_* / daily / hourly 在全量上统计,不受分页影响。
     """
     by_hour = granularity == "hour"
 
@@ -200,7 +205,13 @@ def ad_revenue_report(
             "reward_detail": _reward_detail(row),
         })
 
-    events.sort(key=lambda e: (e["report_date"], e["user_id"], e["created_at"]))
+    # 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
+    # feed_scene 仅信息流 / Draw 有值,激励视频与旧数据为 None;选中后只保留该场景事件。
+    if feed_scene is not None:
+        events = [e for e in events if e.get("feed_scene") == feed_scene]
+
+    # 按时间倒序(新→旧):跨天时新日期在前,同日内新事件在前。
+    events.sort(key=lambda e: (e["report_date"], e["created_at"]), reverse=True)
 
     # 补手机号(admin 展示用,完整不脱敏,与用户 / 钱包 / 比价记录页一致):批量一次查,避免 N+1。
     uids = {e["user_id"] for e in events}
@@ -238,14 +249,38 @@ def ad_revenue_report(
         for d in sorted(daily_map.values(), key=lambda x: x["date"])
     ]
 
+    # 按小时汇总(全量,不受分页 limit/offset 影响):供前端按小时趋势图(单日 granularity=hour 时用)。
+    # 只在 by_hour 下聚合(此时每个 event 带 hour);否则空。前端按天趋势仍用 daily。
+    hourly: list[dict] = []
+    if by_hour:
+        hour_map: dict[int, dict] = {}
+        for e in events:
+            h = e["hour"]
+            if h is None:
+                continue
+            hd = hour_map.get(h)
+            if hd is None:
+                hd = {"hour": h, "impressions": 0, "revenue_yuan": 0.0,
+                      "expected_coin": 0, "actual_coin": 0}
+                hour_map[h] = hd
+            hd["impressions"] += e["impressions"]
+            hd["revenue_yuan"] += e["revenue_yuan"]
+            hd["expected_coin"] += e["expected_coin"]
+            hd["actual_coin"] += e["actual_coin"]
+        hourly = [
+            {**hd, "revenue_yuan": round(hd["revenue_yuan"], 6)}
+            for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
+        ]
+
     return {
         "total": len(events),
-        "truncated": len(events) > limit,
+        "truncated": len(events) > offset + limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
         "total_expected_coin": total_expected_coin,
         "total_actual_coin": total_actual_coin,
         "mismatch_count": mismatch_count,
         "daily": daily,
-        "items": events[:limit],
+        "hourly": hourly,
+        "items": events[offset:offset + limit],
     }

app/admin/routers/ad_revenue.py

@@ -11,7 +11,12 @@ from fastapi import APIRouter, Depends, HTTPException, Query
 
 from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import ad_revenue
-from app.admin.schemas.ad_revenue import AdRevenueDaily, AdRevenueReportOut, AdRevenueRow
+from app.admin.schemas.ad_revenue import (
+    AdRevenueDaily,
+    AdRevenueHourly,
+    AdRevenueReportOut,
+    AdRevenueRow,
+)
 from app.core.rewards import cn_today
 
 router = APIRouter(
@@ -43,10 +48,18 @@ def get_ad_revenue_report(
         str | None,
         Query(description="reward_video / feed / draw;不传=全部类型"),
     ] = None,
+    feed_scene: Annotated[
+        str | None,
+        Query(
+            description="comparison(比价) / coupon(领券) / welfare(福利);不传=全部场景。"
+            "全局筛选,同时影响明细 / 合计 / 趋势"
+        ),
+    ] = None,
     granularity: Annotated[
         str, Query(description="day=按天 / hour=按小时(北京时间);区间>1 天建议用 day")
     ] = "day",
-    limit: Annotated[int, Query(ge=1, le=1000)] = 500,
+    limit: Annotated[int, Query(ge=1, le=1000, description="每页条数(分页大小)")] = 500,
+    offset: Annotated[int, Query(ge=0, description="分页偏移(已跳过的条数)=(页码-1)×每页条数")] = 0,
 ) -> AdRevenueReportOut:
     today = cn_today()
     d_from = _parse_day(date_from, field="date_from", default=today)
@@ -58,12 +71,14 @@ def get_ad_revenue_report(
 
     result = ad_revenue.ad_revenue_report(
         db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
-        user_id=user_id, ad_type=ad_type, granularity=granularity, limit=limit,
+        user_id=user_id, ad_type=ad_type, feed_scene=feed_scene,
+        granularity=granularity, limit=limit, offset=offset,
     )
     return AdRevenueReportOut(
         date_from=d_from.isoformat(),
         date_to=d_to.isoformat(),
         daily=[AdRevenueDaily(**d) for d in result["daily"]],
+        hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
         total=result["total"],
         truncated=result["truncated"],
         total_impressions=result["total_impressions"],

app/admin/schemas/ad_revenue.py

@@ -40,7 +40,7 @@ class AdRevenueRecord(BaseModel):
 
 
 class AdRevenueDaily(BaseModel):
-    """按日期汇总的一天(供前端按天趋势图;全量,不受 limit 影响)。"""
+    """按日期汇总的一天(供前端按天趋势图;全量,不受分页影响)。"""
 
     date: str = Field(..., description="北京时间 YYYY-MM-DD")
     impressions: int = Field(..., description="当天展示条数合计")
@@ -49,6 +49,16 @@ class AdRevenueDaily(BaseModel):
     actual_coin: int = Field(..., description="当天实发金币合计")
 
 
+class AdRevenueHourly(BaseModel):
+    """按北京小时(0–23)汇总的一小时(供前端按小时趋势图;全量,不受分页影响,单日 granularity=hour 时非空)。"""
+
+    hour: int = Field(..., description="北京时间小时 0–23")
+    impressions: int = Field(..., description="该小时展示条数合计")
+    revenue_yuan: float = Field(..., description="该小时预估收益合计(元)")
+    expected_coin: int = Field(..., description="该小时应发金币合计")
+    actual_coin: int = Field(..., description="该小时实发金币合计")
+
+
 class AdRevenueRow(BaseModel):
     """一次广告事件(逐条一行):激励视频展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行。"""
 
@@ -91,8 +101,12 @@ class AdRevenueReportOut(BaseModel):
     date_from: str = Field(..., description="报表起始日期(北京时间 YYYY-MM-DD)")
     date_to: str = Field(..., description="报表结束日期(北京时间 YYYY-MM-DD,闭区间;单日时与 date_from 相同)")
     daily: list[AdRevenueDaily] = Field(..., description="按日期汇总序列(全量,供按天趋势图)")
-    total: int = Field(..., description="广告事件总数(全量,不受 limit 影响)")
-    truncated: bool = Field(..., description="明细是否被 limit 截断")
+    hourly: list[AdRevenueHourly] = Field(
+        default_factory=list,
+        description="按小时汇总序列(全量,供按小时趋势图;按天查询时为空)",
+    )
+    total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
+    truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
     total_impressions: int = Field(..., description="全量展示条数合计")
     total_revenue_yuan: float = Field(..., description="全量收益合计(元)")
     total_expected_coin: int = Field(..., description="全量应发金币合计")

docs/api/admin-ad-revenue-report.md

@@ -27,8 +27,10 @@
   | `date_to` | string | =`date_from` | 结束日 北京时间 `YYYY-MM-DD`,**闭区间**;单日时与 `date_from` 相同 |
   | `user_id` | int | 全部 | 只看某用户;不传=所有用户 |
   | `ad_type` | string | 全部 | `reward_video` / `feed` / `draw`;不传=全部类型 |
+  | `feed_scene` | string | 全部 | `comparison`(比价)/ `coupon`(领券)/ `welfare`(福利);**全局筛选**,同时作用于明细 / 合计 / `daily`·`hourly` 趋势;不传=全部场景 |
   | `granularity` | string | `day` | `day`=按天 / `hour`=按小时(聚合键再加北京时间小时 0–23);**区间>1 天建议用 day** |
-  | `limit` | int(1~1000) | 500 | **展示**明细组数(截断;`total`/`total_*`/`daily` 按全量统计不受影响) |
+  | `limit` | int(1~1000) | 500 | **每页条数**(分页大小);`total`/`total_*`/`daily`/`hourly` 按全量统计不受分页影响 |
+  | `offset` | int(≥0) | 0 | 分页偏移(已跳过条数)=(页码−1)×`limit` |
 
   约束:`date_to` 不早于 `date_from`、区间最长 **92 天**、日期须 `YYYY-MM-DD`,否则 `422`。
 
@@ -36,15 +38,16 @@
   | 字段 | 类型 | 说明 |
   |---|---|---|
   | `date_from` / `date_to` | string | 报表起止日期(闭区间) |
-  | `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受 `limit` 影响) |
-  | `total` | int | 聚合组**总数**(全量,不受 `limit` 影响) |
-  | `truncated` | bool | 明细是否被 `limit` 截断 |
+  | `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受分页影响) |
+  | `hourly` | `AdRevenueHourly[]` | 按小时汇总序列(全量,供按小时趋势图;**仅 `granularity=hour` 时非空**;不受分页影响) |
+  | `total` | int | 当前筛选下的**分页总条数**(全量,不受分页影响;= 前端分页器 total) |
+  | `truncated` | bool | 当前页之后是否还有更多事件(`len(events) > offset + limit`) |
   | `total_impressions` | int | 全量展示条数合计 |
   | `total_revenue_yuan` | float | 全量收益合计(元) |
   | `total_expected_coin` | int | 全量应发金币合计 |
   | `total_actual_coin` | int | 全量实发金币合计 |
   | `mismatch_count` | int | 应发≠实发的组数(=0 说明全部按公式发放) |
-  | `items` | `AdRevenueRow[]` | 聚合明细(按 日期→用户→类型→代码位 排序) |
+  | `items` | `AdRevenueRow[]` | 逐条广告事件(**按时间倒序:新→旧**);`limit`/`offset` 对全量做分页切片,返回当前页 |
 
 ### AdRevenueDaily(`daily[]` — 按天趋势)
 | 字段 | 类型 | 说明 |
@@ -55,6 +58,15 @@
 | `expected_coin` | int | 当天应发金币合计 |
 | `actual_coin` | int | 当天实发金币合计 |
 
+### AdRevenueHourly(`hourly[]` — 按小时趋势,仅 `granularity=hour` 时非空)
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `hour` | int | 北京时间小时 0–23 |
+| `impressions` | int | 该小时展示条数合计 |
+| `revenue_yuan` | float | 该小时预估收益合计(元) |
+| `expected_coin` | int | 该小时应发金币合计 |
+| `actual_coin` | int | 该小时实发金币合计 |
+
 ### AdRevenueRow(`items[]`)
 | 字段 | 类型 | 说明 |
 |---|---|---|