优化：修正广告收益看板统计与明细口径

alembic/versions/guide_video_scene_unique.py

@@ -1,31 +0,0 @@
-"""guide video play count is independent for coupon and comparison
-
-Revision ID: guide_video_scene_unique
-Revises: d8dd2106e438, risk_monitor_generic
-"""
-from alembic import op
-
-revision = "guide_video_scene_unique"
-down_revision = ("d8dd2106e438", "risk_monitor_generic")
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
-    op.create_index(
-        "uq_guide_video_play_user_scene_seq",
-        "guide_video_play",
-        ["user_id", "scene", "seq"],
-        unique=True,
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
-    op.create_index(
-        "uq_guide_video_play_user_seq",
-        "guide_video_play",
-        ["user_id", "seq"],
-        unique=True,
-    )

app/admin/repositories/ad_audit.py

@@ -57,11 +57,12 @@ def _reward_video_rows(
         stmt = stmt.where(AdRewardRecord.user_id == user_id)
 
     records = list(db.execute(stmt).scalars())
-    # S2S 发奖回调不携带实际填充 ADN；用相同用户和 ad_session_id 的展示记录回填。
+    # S2S 发奖回调本身不携带实际填充的 ADN/底层 rit；按客户端在展示时上报的
-    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
+    # ad_session_id 回填。这样“纯发奖”行也能在运营后台追溯到真实广告网络。
+    session_ids = {rec.ad_session_id for rec in records if rec.ad_session_id}
     impression_by_session = {
-        (record.user_id, record.ad_session_id): record
+        (rec.user_id, rec.ad_session_id): rec
-        for record in db.execute(
+        for rec in db.execute(
             select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
         ).scalars()
     } if session_ids else {}
@@ -171,7 +172,7 @@ def _nonblank(value: str | None) -> str | None:
 
 
 def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | None]:
-    """仅在候选展示记录指向唯一 ADN 时回填来源，避免错误归因。"""
+    """仅在候选展示记录指向唯一 ADN 时回填来源，绝不把一次多广告流程猜成某一个网络。"""
     adns = {_nonblank(record.adn) for record in records}
     adns.discard(None)
     if len(adns) != 1:
@@ -184,11 +185,14 @@ def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | No
 
 def _feed_source_fallbacks(
     db: Session, records: list[AdFeedRewardRecord]
-) -> tuple[
+) -> tuple[dict[tuple[int, str], tuple[str | None, str | None]], dict[tuple[int, str, str], tuple[str | None, str | None]]]:
-    dict[tuple[int, str], tuple[str | None, str | None]],
+    """构建信息流来源回填索引。
-    dict[tuple[int, str, str], tuple[str | None, str | None]],
+
-]:
+    新客户端会把 ADN 直接随 feed-reward 上报；旧记录可能缺失。展示收益记录的
-    """为旧信息流发奖记录构建安全来源索引。"""
+    ``ad_session_id`` 是每条 impressionId，而发奖记录保留的是整场会话 ID，因此先按
+    会话精确匹配；匹配不到时仅允许按 ``user + trace_id + 原始 eCPM`` 回填，且候选 ADN
+    必须唯一。trace 内存在多个网络时保持空值，避免错误归因。
+    """
     session_ids = {record.ad_session_id for record in records if record.ad_session_id}
     trace_ids = {record.trace_id for record in records if record.trace_id}
     if not session_ids and not trace_ids:
@@ -223,7 +227,7 @@ def _feed_source(
     by_session: dict[tuple[int, str], tuple[str | None, str | None]],
     by_trace_ecpm: dict[tuple[int, str, str], tuple[str | None, str | None]],
 ) -> tuple[str | None, str | None]:
-    """返回本条发奖广告的来源；无唯一证据时保留原始空值。"""
+    """取得本条发奖广告的真实来源；无唯一证据时返回原始空值。"""
     adn, slot_id = _nonblank(record.adn), _nonblank(record.slot_id)
     if adn and slot_id:
         return adn, slot_id

app/admin/repositories/ad_revenue.py

@@ -22,7 +22,7 @@ report_date / reward_date 归日。
 """
 from __future__ import annotations
 
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime, time, timedelta
 from datetime import date as _date
 
 from sqlalchemy import select
@@ -81,9 +81,9 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 # ad_feed_reward_record,由 audit 内部按 ad_type 区分(feed 含历史 NULL,draw 仅 ad_type=="draw")。
 _AUDIT_SCENES = {"reward_video", "feed", "draw"}
 
-# 激励视频未满足有效播放条件时不计客户端预估收益。客户端仍会在 onAdShow
+# GroMore 官方说明第三方 ADN 的 Reporting API 最晚约 13:50 更新。只有 D+1 14:00
-# 上报 eCPM，随后才在关闭时补报以下终态，因此必须在展示/发奖合并后修正收益。
+# 之后完成的同步才标记为「API 同步窗口完成」；这不代表覆盖全部 ADN 或最终结算。
-_ZERO_REVENUE_REWARD_VIDEO_STATUSES = frozenset({"closed_early", "too_short"})
+_PANGLE_API_FINAL_SYNC_TIME = time(hour=14)
 
 
 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
@@ -96,13 +96,34 @@ _REWARD_DETAIL_KEYS = (
 
 def _reward_detail(row: dict) -> dict:
     """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
-    detail = {key: row[key] for key in _REWARD_DETAIL_KEYS}
+    detail = {k: row[k] for k in _REWARD_DETAIL_KEYS}
-    # 聚合父行可能包含多个 ADN，来源必须保留在每一条发奖明细上。
+    # 发奖明细必须保留自己的广告网络，不能复用整场聚合父行的来源：
+    # 同一次比价/领券可能先后由不同 ADN 填充。
     detail["adn"] = row.get("adn")
     detail["slot_id"] = row.get("slot_id")
     return detail
 
 
+def _as_cn(dt: datetime) -> datetime:
+    """数据库 synced_at → 北京时间；SQLite naive 值按 UTC 处理。"""
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return dt.astimezone(rewards.CN_TZ)
+
+
+def _pangle_api_day_complete(day: str, aggregate: dict) -> bool:
+    """某天 API 收益是否已在 D+1 14:00 后同步（仅表示同步窗口完成）。"""
+    synced_at = aggregate.get("synced_at")
+    if aggregate.get("api_revenue_yuan") is None or synced_at is None:
+        return False
+    cutoff = datetime.combine(
+        _date.fromisoformat(day) + timedelta(days=1),
+        _PANGLE_API_FINAL_SYNC_TIME,
+        tzinfo=rewards.CN_TZ,
+    )
+    return _as_cn(synced_at) >= cutoff
+
+
 def ad_revenue_report(
     db: Session,
     *,
@@ -190,12 +211,10 @@ def ad_revenue_report(
             "has_impression": True,
             "impressions": 1,
             "ecpm": rec.ecpm_raw,
-            # 单次展示收益(元)= eCPM元 ÷ 1000(每千次→单次)。eCPM 先钳到 AD_ECPM_MAX_FEN(¥500 CPM)
+            # 客户端 SDK 展示预估收益(元)= 后端留存 getEcpm 元/千次 ÷ 1000。
-            # 再折收益,与发奖口径 [rewards.calculate_ad_reward_coin] 一致(2026-06-29 修:原裸 parse_ecpm_yuan
+            # 这里不能复用发奖防作弊的 ¥500 CPM 钳顶：钳顶只限制金币成本，不改变广告已产生的
-            # 不钳,伪造/异常天价 eCPM 会把报表预估收益冲到任意大;金币侧已钳、收益侧漏钳)。
+            # 收入估值。onAdShow 已发生即计展示收入，是否看满只影响发奖，不影响广告收入。
-            "revenue_yuan": round(
+            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0, 6),
-                min(rewards.parse_ecpm_yuan(rec.ecpm_raw), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0, 6,
-            ),
             "adn": rec.adn,
             "slot_id": rec.slot_id,
             "sub_rewards": [],
@@ -210,11 +229,6 @@ def ad_revenue_report(
                 "matched": bool(rwd["matched"]),
                 "reward_detail": _reward_detail(rwd),
             })
-            if (
-                rec.ad_type == "reward_video"
-                and rwd["status"] in _ZERO_REVENUE_REWARD_VIDEO_STATUSES
-            ):
-                ev["revenue_yuan"] = 0.0
         else:
             # 纯展示(信息流逐条展示、激励视频缺发奖记录):不计对账,matched=True。
             ev.update({
@@ -275,10 +289,10 @@ def ad_revenue_report(
         # 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条
         ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")]
         avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm")
-        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进
+        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算)。只放进
         # row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。
         row_revenue = round(sum(
-            min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
+            rewards.parse_ecpm_yuan(g["ecpm"]) / 1000.0
             for g in group if g.get("ecpm")
         ), 6)
         events.append({
@@ -368,13 +382,15 @@ def ad_revenue_report(
         for d in sorted(daily_map.values(), key=lambda x: x["date"])
     ]
 
-    # 穿山甲后台收益(GroMore 数据 API,T+1 入库 ad_pangle_daily_revenue):汇总 + 按天趋势级展示,
+    # GroMore 排序价预估 / ADN Reporting API 收益(T+1 入库):汇总 + 按天趋势级展示,
     # 与上面客户端自报 eCPM 折算的预估并列对照(看 gap)。穿山甲数据**无用户/场景/类型维度**,故仅在
     # 「全量视图」(未按 user_id / ad_type / feed_scene 过滤)给值;一旦带这些过滤,穿山甲数无法对应口径
     # → 置 None,前端显示「-」并提示。逐条事件行不动(仍是客户端预估)。
     pangle_filterable = user_id is None and ad_type is None and feed_scene is None
     total_pangle_revenue_yuan: float | None = None
     total_pangle_api_revenue_yuan: float | None = None
+    pangle_api_revenue_complete = False
+    pangle_latest_synced_at: datetime | None = None
     if pangle_filterable:
         pangle_aggs = ad_pangle_revenue.aggregate_by_date(
             db,
@@ -392,6 +408,12 @@ def ad_revenue_report(
             total_pangle_revenue_yuan = round(sum(a["revenue_yuan"] for a in pangle_aggs), 6)
             api_vals = [a["api_revenue_yuan"] for a in pangle_aggs if a["api_revenue_yuan"] is not None]
             total_pangle_api_revenue_yuan = round(sum(api_vals), 6) if api_vals else None
+            sync_times = [a["synced_at"] for a in pangle_aggs if a["synced_at"] is not None]
+            pangle_latest_synced_at = max(sync_times) if sync_times else None
+            pangle_api_revenue_complete = all(
+                day in by_date and _pangle_api_day_complete(day, by_date[day])
+                for day in _date_range(date_from, date_to)
+            )
 
     # 按小时汇总(全量,不受分页 limit/offset 影响):供前端按小时趋势图(单日 granularity=hour 时用)。
     # 只在 by_hour 下聚合(此时每个 event 带 hour);否则空。前端按天趋势仍用 daily。
@@ -416,41 +438,50 @@ def ad_revenue_report(
             for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
         ]
 
-    # 分广告类型小计(按 ad_type:展示条数 + 预估收益;eCPM 由前端用 收益÷展示×1000 算)。
+    def _aggregate_stats(bucket_of) -> dict[str, dict]:
-    # 基于全量(已按 feed_scene 过滤)events;前端只取 draw / reward_video 两类展示。
+        """按展示事件聚合收益 / 加权 SDK eCPM，避免前端漏合并历史类型。"""
-    type_map: dict[str, dict] = {}
+        stat_map: dict[str, dict] = {}
-    for e in events:
+        for e in events:
-        t = type_map.get(e["ad_type"])
+            bucket = bucket_of(e)
-        if t is None:
+            if bucket is None:
-            t = {"impressions": 0, "revenue_yuan": 0.0}
+                continue
-            type_map[e["ad_type"]] = t
+            stat = stat_map.setdefault(bucket, {
-        t["impressions"] += e["impressions"]
+                "impressions": 0,
-        t["revenue_yuan"] += e["revenue_yuan"]
+                "revenue_yuan": 0.0,
-    type_stats = {
+                "ecpm_fen_sum": 0.0,
-        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
+            })
-        for k, v in type_map.items()
+            impressions = int(e["impressions"])
-    }
+            stat["impressions"] += impressions
+            stat["revenue_yuan"] += e["revenue_yuan"]
+            # eCPM 必须以每次真实展示为权重；纯发奖父行 impressions=0，不能参与分母或均值。
+            stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(e["ecpm"]) * impressions
+        return {
+            key: {
+                "impressions": value["impressions"],
+                "revenue_yuan": round(value["revenue_yuan"], 6),
+                "ecpm_yuan": round(
+                    value["ecpm_fen_sum"] / value["impressions"] / 100.0,
+                    6,
+                ) if value["impressions"] else 0.0,
+            }
+            for key, value in stat_map.items()
+        }
 
-    # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
+    # 原始 ad_type 小计，供明细筛选和排查使用。
-    # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
+    type_stats = _aggregate_stats(lambda e: e["ad_type"])
-    # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
+    # 经营看板使用的规范分类：Draw 包含历史 feed；看视频包含福利与提现视频。
-    # 改为服务端在全量上聚合下发(也顺带不受 limit 分页截断影响)。feed_scene 为空(激励视频 /
+    # 这两个集合与筛选逻辑保持一致，避免只取 draw / reward_video 而漏算历史或提现数据。
-    # 旧数据)不计入任何场景桶。
+    category_stats = _aggregate_stats(
-    scene_map: dict[str, dict] = {}
+        lambda e: (
-    for e in events:
+            "draw" if e["ad_type"] in {"draw", "feed"}
-        sc = e.get("feed_scene")
+            else "video" if e["ad_type"] in {"reward_video", "withdrawal_video"}
-        if not sc:
+            else None
-            continue
+        )
-        s = scene_map.get(sc)
+    )
-        if s is None:
+
-            s = {"impressions": 0, "revenue_yuan": 0.0}
+    # 分场景小计，同 type_stats 基于全量 events，供数据大盘「领券广告 / 比价广告」卡使用。
-            scene_map[sc] = s
+    # feed_scene 为空的激励视频 / 历史数据不计入任何场景桶。
-        s["impressions"] += e["impressions"]
+    scene_stats = _aggregate_stats(lambda e: e.get("feed_scene"))
-        s["revenue_yuan"] += e["revenue_yuan"]
-    scene_stats = {
-        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
-        for k, v in scene_map.items()
-    }
 
     # DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
     # 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
@@ -475,9 +506,11 @@ def ad_revenue_report(
         "truncated": len(main_rows) > offset + limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
-        # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。
+        # GroMore 排序价预估 + ADN Reporting API 收益;非全量视图或无数据为 None。
         "total_pangle_revenue_yuan": total_pangle_revenue_yuan,
         "total_pangle_api_revenue_yuan": total_pangle_api_revenue_yuan,
+        "pangle_api_revenue_complete": pangle_api_revenue_complete,
+        "pangle_latest_synced_at": pangle_latest_synced_at,
         "pangle_revenue_available": total_pangle_revenue_yuan is not None,
         "total_expected_coin": total_expected_coin,
         "total_actual_coin": total_actual_coin,
@@ -485,6 +518,7 @@ def ad_revenue_report(
         "daily": daily,
         "hourly": hourly,
         "type_stats": type_stats,
+        "category_stats": category_stats,
         "scene_stats": scene_stats,
         "dau": dau,
         "items": main_rows[offset:offset + limit],

app/admin/repositories/queries.py

@@ -1228,11 +1228,6 @@ def _cn_wall_to_utc(dt: datetime) -> datetime:
     return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
 
 
-def _coin_record_sort_key(row: dict) -> datetime:
-    """金币明细跨数据源排序键：兼容 SQLite naive 与 PostgreSQL aware 时间。"""
-    return _as_utc(row["created_at"])
-
-
 def user_coin_records(
     db: Session,
     user_id: int,
@@ -1324,10 +1319,7 @@ def user_coin_records(
             "coin": rec.amount,
         })
 
-    # SQLite 常返回 naive datetime，PostgreSQL timestamptz 返回 aware datetime；
+    rows.sort(key=lambda r: r["created_at"], reverse=True)
-    # 统一成 aware UTC 排序，避免线上合并广告记录与签到记录时抛
-    # “can't compare offset-naive and offset-aware datetimes”。
-    rows.sort(key=_coin_record_sort_key, reverse=True)
     has_more = len(rows) > offset + limit
 
     # 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)

app/admin/routers/ad_revenue.py

@@ -99,6 +99,7 @@ def get_ad_revenue_report(
         daily=[AdRevenueDaily(**d) for d in result["daily"]],
         hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
         type_stats={k: AdRevenueTypeStat(**v) for k, v in result["type_stats"].items()},
+        category_stats={k: AdRevenueTypeStat(**v) for k, v in result["category_stats"].items()},
         scene_stats={k: AdRevenueTypeStat(**v) for k, v in result["scene_stats"].items()},
         dau=result["dau"],
         total=result["total"],
@@ -107,6 +108,8 @@ def get_ad_revenue_report(
         total_revenue_yuan=result["total_revenue_yuan"],
         total_pangle_revenue_yuan=result["total_pangle_revenue_yuan"],
         total_pangle_api_revenue_yuan=result["total_pangle_api_revenue_yuan"],
+        pangle_api_revenue_complete=result["pangle_api_revenue_complete"],
+        pangle_latest_synced_at=result["pangle_latest_synced_at"],
         pangle_revenue_available=result["pangle_revenue_available"],
         total_expected_coin=result["total_expected_coin"],
         total_actual_coin=result["total_actual_coin"],

app/admin/routers/guide_video.py

@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
 """
 from __future__ import annotations
 
-from typing import Annotated, Literal
+from typing import Annotated
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 
@@ -27,21 +27,14 @@ router = APIRouter(
 )
 
 
-GuideScene = Literal["coupon", "comparison"]
+def _out(db: AdminDb) -> GuideVideoConfigOut:
-
-
-def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
     """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
-    return GuideVideoConfigOut(
+    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
-        scene=scene,
-        **guide_video.get_config(db, scene),
-        **guide_video.play_stats(db, scene),
-    )
 
 
 @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
-def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
+def get_config(db: AdminDb) -> GuideVideoConfigOut:
-    return _out(db, scene)
+    return _out(db)
 
 
 @router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -50,23 +43,21 @@ def update_config(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     before, after = guide_video.update_config(
         db,
         enabled=body.enabled,
         max_plays=body.max_plays,
         reward_coin=body.reward_coin,
-        scene=scene,
         admin_id=admin.id,
         commit=False,
     )
     write_audit(
         db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
+        detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
-    return _out(db, scene)
+    return _out(db)
 
 
 @router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -74,26 +65,23 @@ async def upload_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    file: Annotated[UploadFile, File()],
+    file: UploadFile = File(...),
-    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     data = await file.read()
     try:
         url = media.save_guide_video(data)
     except media.MediaError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
-    before, after = guide_video.set_video(
+    before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
-        db, url, scene=scene, admin_id=admin.id, commit=False
-    )
     write_audit(
         db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
+        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
         ip=get_client_ip(request), commit=False,
     )
     db.commit()
     # 提交成功后再删旧片,避免新片没落库就把旧片丢了
     media.delete_guide_video(before.get("video_url"))
-    return _out(db, scene)
+    return _out(db)
 
 
 @router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -101,16 +89,13 @@ def delete_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
-    before, after = guide_video.set_video(
+    before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
-        db, None, scene=scene, admin_id=admin.id, commit=False
-    )
     write_audit(
         db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
+        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
     media.delete_guide_video(before.get("video_url"))
-    return _out(db, scene)
+    return _out(db)

app/admin/routers/users.py

@@ -86,7 +86,7 @@ def get_user_reward_stats(
     date_to: Annotated[datetime | None, Query()] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
-    if not user_repo.user_exists(db, user_id):
+    if user_repo.get_user_by_id(db, user_id) is None:
         raise HTTPException(status_code=404, detail="用户不存在")
     return UserRewardStats(
         **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)

app/admin/schemas/ad_revenue.py

@@ -49,12 +49,12 @@ class AdRevenueDaily(BaseModel):
 
     date: str = Field(..., description="北京时间 YYYY-MM-DD")
     impressions: int = Field(..., description="当天展示条数合计")
-    revenue_yuan: float = Field(..., description="当天客户端有效预估收益合计(元;eCPM 折算)")
+    revenue_yuan: float = Field(..., description="当天客户端 SDK 展示预估合计(元;后端留存 eCPM 折算)")
     pangle_revenue_yuan: float | None = Field(
-        None, description="当天穿山甲后台预估收益(元;GroMore revenue);非全量视图/无数据为空"
+        None, description="当天 GroMore 排序价预估(元;revenue,非结算收入);非全量视图/无数据为空"
     )
     pangle_api_revenue_yuan: float | None = Field(
-        None, description="当天穿山甲收益Api(元;GroMore api_revenue,更接近结算);未配/当天/无数据为空"
+        None, description="当天 ADN Reporting API 收益(元;GroMore api_revenue);未配/当天/无数据为空"
     )
     expected_coin: int = Field(..., description="当天应发金币合计")
     actual_coin: int = Field(..., description="当天实发金币合计")
@@ -71,10 +71,11 @@ class AdRevenueHourly(BaseModel):
 
 
 class AdRevenueTypeStat(BaseModel):
-    """按广告类型(ad_type)的小计:展示条数 + 预估收益(eCPM 由前端用 收益÷展示×1000 算)。"""
+    """展示条数、SDK 展示预估收益与按展示次数加权的 SDK eCPM。"""
 
     impressions: int = Field(..., description="该类型展示条数合计")
     revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
+    ecpm_yuan: float = Field(..., description="按展示次数加权的 SDK eCPM(元/千次)")
 
 
 class AdRevenueRow(BaseModel):
@@ -100,15 +101,15 @@ class AdRevenueRow(BaseModel):
     ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
     revenue_yuan: float = Field(
         ...,
-        description="本次有效展示预估收益(元)= eCPM元 ÷ 1000;纯发奖、激励视频提前关闭/时长不足=0",
+        description="本次 SDK 展示预估收益(元)=后端留存 eCPM 元 ÷ 1000;是否满足发奖条件不改变展示收入预估",
     )
     row_revenue_yuan: float | None = Field(
         None,
         description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
         "其它行为空(前端回退取 revenue_yuan)。不进合计/趋势,避免与展示侧重复计",
     )
-    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空")
+    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);历史或未上报展示来源为空")
-    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
+    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);历史或未上报展示来源为空")
     # ── 发奖侧 ──
     has_reward: bool = Field(..., description="是否有发奖记录(激励视频合并行 / 信息流整场发奖行=True;纯展示=False)")
     status: str | None = Field(None, description="发奖状态 granted/closed_early/too_short/…;纯展示为空")
@@ -142,7 +143,11 @@ class AdRevenueReportOut(BaseModel):
     )
     type_stats: dict[str, AdRevenueTypeStat] = Field(
         default_factory=dict,
-        description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
+        description="按原始广告类型(ad_type)小计，供筛选与排查使用",
+    )
+    category_stats: dict[str, AdRevenueTypeStat] = Field(
+        default_factory=dict,
+        description="按经营分类小计：draw=draw+历史 feed，video=reward_video+withdrawal_video",
     )
     scene_stats: dict[str, AdRevenueTypeStat] = Field(
         default_factory=dict,
@@ -158,20 +163,28 @@ class AdRevenueReportOut(BaseModel):
     total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
     truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
     total_impressions: int = Field(..., description="全量展示条数合计")
-    total_revenue_yuan: float = Field(..., description="全量客户端有效预估收益合计(元;eCPM 折算)")
+    total_revenue_yuan: float = Field(..., description="全量客户端 SDK 展示预估合计(元;后端留存 eCPM 折算)")
     total_pangle_revenue_yuan: float | None = Field(
         None,
-        description="全量穿山甲后台预估收益合计(元;GroMore revenue)。穿山甲无用户/类型/场景维度,"
+        description="全量 GroMore 排序价预估合计(元;revenue,非结算收入)。GroMore 无用户/类型/场景维度,"
         "仅「全量视图」(未按 user_id/ad_type/feed_scene 过滤)时有值,否则为 null",
     )
     total_pangle_api_revenue_yuan: float | None = Field(
         None,
-        description="全量穿山甲收益Api合计(元;GroMore api_revenue,各 ADN 回传、更接近结算);"
+        description="全量 ADN Reporting API 收益合计(元;GroMore api_revenue,仅已配置回传的 ADN);"
         "未配 Reporting / 查当天 / 非全量视图 时为 null",
     )
+    pangle_api_revenue_complete: bool = Field(
+        False,
+        description="所选每一天是否都已在 D+1 14:00 后完成 API 同步窗口；不代表覆盖全部 ADN 或最终结算",
+    )
+    pangle_latest_synced_at: datetime | None = Field(
+        None,
+        description="所选范围穿山甲/GroMore 日报最近同步时间",
+    )
     pangle_revenue_available: bool = Field(
         False,
-        description="本次结果是否带穿山甲后台收益(=全量视图且已同步到数据)。false 时前端「穿山甲收益」显示「-」",
+        description="本次结果是否带 GroMore/ADN 收益(=全量视图且已同步到数据)。false 时前端显示「-」",
     )
     total_expected_coin: int = Field(..., description="全量应发金币合计")
     total_actual_coin: int = Field(..., description="全量实发金币合计")

app/admin/schemas/guide_video.py

@@ -7,7 +7,6 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
 
 
 class GuideVideoConfigOut(BaseModel):
-    scene: str
     enabled: bool
     video_url: str | None = None   # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
     max_plays: int

app/api/v1/ad.py

@@ -289,7 +289,8 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     """客户端在广告展示后(onAdShow 读 getShowEcpm)上报 eCPM,落库做内部收益统计/对账。
 
     Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
-    丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
+    丢一两条不影响发奖业务(收入另由 ADN Reporting API 对账)。eCPM 与发奖(S2S)是两条独立流,
+    不逐条关联。
     """
     attributed_trace_id = crud_ecpm.attributable_trace_id(
         db,

app/core/config.py

@@ -318,7 +318,7 @@ class Settings(BaseSettings):
     # ===== 穿山甲 GroMore 数据 API(报表收益拉取,T+1)=====
     # ⚠️ 与上面发奖回调的 m-key 是【两套完全不同的凭证】:这三样在穿山甲后台
     # 「接入中心 → GroMore-API → 聚合数据报告 API」文档页领取(user_id / role_id / Security Key),
-    # 仅用于按天拉 GroMore 收益报表(revenue 预估收益 + api_revenue 收益Api),不参与发奖。
+    # 仅用于按天拉 GroMore 报表(revenue 排序价预估 + api_revenue ADN Reporting 收益),不参与发奖。
     # 该 API 只能查【GroMore 聚合代码位】的数据(=我们 useMediation 的口径),非穿山甲 SDK 数据;
     # 且不提供用户/设备维度(官方明确),故收益只能落到 日期×代码位 汇总,不能挂到逐条事件。
     # 子账号(role_id≠user_id)需主账号在「角色管理」授予「查看全部数据」权限,否则查不到

app/integrations/pangle_report.py

@@ -14,8 +14,8 @@
   - 只返回【GroMore 聚合代码位】在 GroMore 内的数据(=我们 useMediation 的口径),
     查不到穿山甲 SDK 自身的数据;
   - **不提供分用户/设备维度**(官方 FAQ 明确拒绝),最细到 日期×应用×代码位×广告源;
-  - `revenue` = 预估收益(元,所有 ADN 都有);`api_revenue` = 收益Api(各 ADN 经 Reporting
+  - `revenue` = 排序价/竞价实时价预估(元,非结算收入);`api_revenue` = 各 ADN 经 Reporting
-    回传、按实时汇率折算账号币种,更接近结算),需后台为该 ADN 配置 Reporting 才有、且不支持当天;
+    回传、按实时汇率折算账号币种的收益,需后台为该 ADN 配置 Reporting 才有、且不支持当天;
   - 「今天」与「今天以前」必须分开查;天级跨度 ≤ 1 个月、不早于 12 个月。
 """
 from __future__ import annotations

app/models/ad_pangle_revenue.py

@@ -1,15 +1,15 @@
-"""穿山甲 GroMore 天级收益报表(后台结算口径,定时拉取入库)。
+"""GroMore 天级排序价预估与 ADN Reporting 收益(定时拉取入库)。
 
 每行 = GroMore 数据 API 返回的一条「日期 × 应用 × 代码位」聚合收益(`integrations/pangle_report`
-+ `scripts/sync_pangle_revenue` 落库)。**权威/预估收益的来源**,与 `ad_ecpm_record`(客户端自报
++ `scripts/sync_pangle_revenue` 落库)，与 `ad_ecpm_record`(客户端 SDK eCPM 折算的预估)
-eCPM 折算的预估)互为对照:
+互为对照:
 
-- `revenue_yuan` ← 接口 `revenue`(预估收益,元;排序价×展示/1000,所有 ADN 都有);
+- `revenue_yuan` ← 接口 `revenue`(排序价/竞价实时价预估,元,不是结算收入);
-- `api_revenue_yuan` ← 接口 `api_revenue`(收益Api,元;各 ADN 经 Reporting 回传、更接近结算;
+- `api_revenue_yuan` ← 接口 `api_revenue`(各 ADN Reporting 回传收益,元,更接近结算;
   未配置该 ADN 的 Reporting 或查当天时为空)。
 
 ⚠️ 穿山甲不提供分用户/设备维度,故本表最细只到 日期×应用×代码位,**无法挂到逐条广告事件**;
-广告收益报表里只用于汇总/趋势级的「穿山甲后台收益」,不改逐条行的客户端预估。
+广告收益报表里只用于汇总/趋势级的 GroMore/ADN 对账,不改逐条行的客户端预估。
 """
 from __future__ import annotations
 
@@ -51,9 +51,9 @@ class AdPangleDailyRevenue(Base):
     our_code_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
     # 广告源(接口 network 数字→名,如 pangle/gdt);"" = 未分广告源的代码位汇总行(当前默认口径)。
     adn: Mapped[str] = mapped_column(String(16), nullable=False, default="")
-    # 预估收益(元)← 接口 revenue。
+    # 排序价/竞价实时价预估(元)← 接口 revenue，非结算收入。
     revenue_yuan: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
-    # 收益Api(元)← 接口 api_revenue;未配 Reporting / 当天 等情况接口不返回 → NULL。
+    # ADN Reporting API 收益(元)← api_revenue;未配 Reporting / 当天等情况不返回 → NULL。
     api_revenue_yuan: Mapped[float | None] = mapped_column(Float, nullable=True)
     # 预估 eCPM 原值(接口 ecpm,单位元/千次,**与客户端 getEcpm 的「分」不同**),参考用原样存。
     ecpm: Mapped[str | None] = mapped_column(String(32), nullable=True)

app/models/guide_video.py

@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
         # start_play 捕获 IntegrityError 降级成"这次不放视频"。
         # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
         # 要整表重建),autogenerate 才不会每次报一条假 diff。
-        Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
+        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

app/repositories/ad_ecpm.py

@@ -1,8 +1,8 @@
 """广告 eCPM 上报 CRUD(内部收益统计/对账)。
 
 客户端在广告展示后(onAdShow)读到 eCPM,经鉴权接口上报,这里落库。鉴权接口已确保
-user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响业务,
+user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响发奖业务;
-穿山甲后台报表是结算权威兜底。
+汇总收入以 ADN Reporting API 和最终结算单为准。
 """
 from __future__ import annotations
 
@@ -96,7 +96,7 @@ def create_ecpm_record(
         db.rollback()
         # 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
         # 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
-        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
+        # (收入另由 ADN Reporting API 对账),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
         # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
         # 旧逻辑在此 raise 成 500(本应静默吞掉)。
         existing = _find_by_session_global(db, ad_session_id)

app/repositories/ad_pangle_revenue.py

@@ -1,12 +1,13 @@
 """穿山甲 GroMore 天级收益 读写(`ad_pangle_daily_revenue` 表)。
 
 `scripts/sync_pangle_revenue` 拉数后调 `upsert_daily_rows` 落库(同一(日期×应用×代码位×广告源)
-幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取「穿山甲后台收益」做
+幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取 GroMore/ADN 收益做
 汇总/趋势级展示。穿山甲无用户维度,故这里不涉及 user_id。
 """
 from __future__ import annotations
 
 from collections.abc import Collection
+from datetime import datetime
 from typing import Any, TypedDict
 
 from sqlalchemy import func, select
@@ -23,6 +24,7 @@ class PangleDateAgg(TypedDict):
     revenue_yuan: float
     api_revenue_yuan: float | None
     impressions: int
+    synced_at: datetime | None
 
 
 def upsert_daily_rows(db: Session, rows: list[dict[str, Any]]) -> dict[str, int]:
@@ -87,6 +89,7 @@ def aggregate_by_date(
             func.sum(AdPangleDailyRevenue.revenue_yuan),
             func.sum(AdPangleDailyRevenue.api_revenue_yuan),
             func.sum(AdPangleDailyRevenue.impressions),
+            func.max(AdPangleDailyRevenue.synced_at),
         )
         .where(
             AdPangleDailyRevenue.report_date >= date_from,
@@ -103,11 +106,12 @@ def aggregate_by_date(
         stmt = stmt.where(AdPangleDailyRevenue.our_code_id.in_(our_code_ids))
 
     out: list[PangleDateAgg] = []
-    for report_date, rev, api_rev, imp in db.execute(stmt).all():
+    for report_date, rev, api_rev, imp, synced_at in db.execute(stmt).all():
         out.append(PangleDateAgg(
             date=report_date,
             revenue_yuan=round(float(rev or 0.0), 6),
             api_revenue_yuan=(round(float(api_rev), 6) if api_rev is not None else None),
             impressions=int(imp or 0),
+            synced_at=synced_at,
         ))
     return out

app/repositories/comparison.py

@@ -55,22 +55,13 @@ def _product_names_from_items(items: list | None) -> str | None:
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
-    _pr = payload.platform_results or {}
 
-    def _is_short(r) -> bool:
+    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
-        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
-        # platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
-        info = _pr.get(r.platform_id) if r.platform_id else None
-        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
-
-    # 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
-    # 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
-    priced = [r for r in results if r.price is not None]
-    clean = [r for r in priced if not _is_short(r)]
     best = None
-    if clean:
+    priced = [r for r in results if r.price is not None]
+    if priced:
         best = min(
-            clean,
+            priced,
             key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
         )
 
@@ -205,29 +196,14 @@ def upsert_record(
 # ============================================================
 
 
-def _derive_from_results(
+def _derive_from_results(results: list[dict]) -> dict:
-    results: list[dict], platform_results: dict | None = None
-) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
-
-    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
-    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
-    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
-    _pr = platform_results or {}
-
-    def _is_short(r: dict) -> bool:
-        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
-        pid = r.get("platform_id")
-        info = _pr.get(pid) if pid else None
-        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
-
     priced = [r for r in results if r.get("price") is not None]
-    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
     best = None
-    if clean:
+    if priced:
         best = min(
-            clean,
+            priced,
             key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
         )
     src_row = next((r for r in results if r.get("is_source")), None)
@@ -413,7 +389,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
+    derived = _derive_from_results(results)
     # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
     items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(

app/repositories/guide_video.py

@@ -30,11 +30,7 @@ from app.models.app_config import AppConfig
 from app.models.guide_video import GuideVideoPlay
 from app.repositories import wallet as crud_wallet
 
-SCENES = ("coupon", "comparison")
+_KEY = "coupon_guide_video"
-_KEY_BY_SCENE = {
-    "coupon": "coupon_guide_video",
-    "comparison": "comparison_guide_video",
-}
 
 #: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
 BIZ_TYPE = "guide_video"
@@ -45,7 +41,7 @@ _DEFAULTS: dict[str, Any] = {
     "enabled": True,
     "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
     "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
-    "reward_coin": 100,  # 每次固定金币
+    "reward_coin": 120,  # 每次固定金币
 }
 
 _FIELDS = tuple(_DEFAULTS.keys())
@@ -69,28 +65,19 @@ def _merge(raw: Any) -> dict[str, Any]:
     return out
 
 
-def _config_key(scene: str) -> str:
+def get_config(db: Session) -> dict[str, Any]:
-    if scene not in _KEY_BY_SCENE:
-        raise ValueError(f"unsupported guide video scene: {scene}")
-    return _KEY_BY_SCENE[scene]
-
-
-def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
     """完整配置 + updated_at(admin 读 / 业务读共用)。"""
-    row = db.get(AppConfig, _config_key(scene))
+    row = db.get(AppConfig, _KEY)
     cfg = _merge(row.value if row is not None else None)
     cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
     return cfg
 
 
-def _write(
+def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
-    db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
-) -> dict[str, Any]:
     """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
-    key = _config_key(scene)
+    row = db.get(AppConfig, _KEY)
-    row = db.get(AppConfig, key)
     if row is None:
-        row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
+        row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
         db.add(row)
     else:
         row.value = value  # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -111,12 +98,11 @@ def update_config(
     enabled: bool | None = None,
     max_plays: int | None = None,
     reward_coin: int | None = None,
-    scene: str = "coupon",
     admin_id: int,
     commit: bool = True,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
-    row = db.get(AppConfig, _config_key(scene))
+    row = db.get(AppConfig, _KEY)
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     if enabled is not None:
@@ -125,52 +111,45 @@ def update_config(
         new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
     if reward_coin is not None:
         new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
-    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
+    after = _write(db, new_value, admin_id=admin_id, commit=commit)
     return before, after
 
 
 def set_video(
-    db: Session, video_url: str | None, *, scene: str = "coupon",
+    db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
-    admin_id: int, commit: bool = True
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
-    row = db.get(AppConfig, _config_key(scene))
+    row = db.get(AppConfig, _KEY)
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     new_value["video_url"] = video_url
-    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
+    after = _write(db, new_value, admin_id=admin_id, commit=commit)
     return before, after
 
 
 # ===== 播放计次 =====
 
 
-def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
+def used_plays(db: Session, user_id: int) -> int:
     """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
     return int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.user_id == user_id,
+                GuideVideoPlay.user_id == user_id
-                GuideVideoPlay.scene == scene,
             )
         ).scalar_one()
     )
 
 
-def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
+def play_stats(db: Session) -> dict[str, int]:
     """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
     total = int(
-        db.execute(
+        db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
-            select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.scene == scene
-            )
-        ).scalar_one()
     )
     granted = int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.status == "granted",
+                GuideVideoPlay.status == "granted"
-                GuideVideoPlay.scene == scene,
             )
         ).scalar_one()
     )
@@ -189,12 +168,11 @@ def start_play(
       reward_coin  播完/中途关闭都发的固定金币
       seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
     """
-    _config_key(scene)
+    cfg = get_config(db)
-    cfg = get_config(db, scene)
     video_url = (cfg.get("video_url") or "").strip()
     max_plays = int(cfg.get("max_plays") or 0)
     reward_coin = int(cfg.get("reward_coin") or 0)
-    used = used_plays(db, user_id, scene)
+    used = used_plays(db, user_id)
 
     def _miss(used_now: int) -> dict[str, Any]:
         return {
@@ -216,7 +194,7 @@ def start_play(
         scene=scene,
         seq=seq,
         video_url=video_url,
-        coin=reward_coin,
+        coin=0,
         status="playing",
         completed=0,
         started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -232,7 +210,7 @@ def start_play(
             db.flush()
     except IntegrityError:
         db.rollback()
-        return _miss(used_plays(db, user_id, scene))
+        return _miss(used_plays(db, user_id))
     return {
         "should_play": True,
         "video_url": video_url,
@@ -263,8 +241,7 @@ def grant_play(
     """
     token = (play_token or "").strip()
     # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
-    play = _find_play(db, user_id, token)
+    coin = int(get_config(db).get("reward_coin") or 0)
-    coin = int(play.coin if play is not None else 0)
 
     # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
     # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',

app/repositories/user.py

@@ -93,11 +93,6 @@ def get_user_by_id(db: Session, user_id: int) -> User | None:
     return db.get(User, user_id)
 
 
-def user_exists(db: Session, user_id: int) -> bool:
-    """只查主键判断用户是否存在，避免只读统计接口依赖完整用户表结构。"""
-    return db.scalar(select(User.id).where(User.id == user_id)) is not None
-
-
 def get_user_by_phone(db: Session, phone: str) -> User | None:
     stmt = select(User).where(User.phone == phone)
     return db.execute(stmt).scalar_one_or_none()

app/schemas/guide_video.py

@@ -1,15 +1,13 @@
 """新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
 from __future__ import annotations
 
-from typing import Literal
-
 from pydantic import BaseModel, Field
 
 
 class GuideVideoStartIn(BaseModel):
     """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
 
-    scene: Literal["coupon", "comparison"] = "coupon"
+    scene: str = Field(default="coupon", max_length=16)
 
 
 class GuideVideoStartOut(BaseModel):

deploy/pangle-revenue.md

@@ -1,13 +1,13 @@
 # 穿山甲 GroMore 收益拉取 定时任务 — 运维手册
 
-> 对象:维护「每天拉穿山甲后台收益入库」这套定时任务的同事。
+> 对象:维护「每天拉 GroMore / ADN 收益入库」这套定时任务的同事。
 > 🔒 服务器登录信息见**私密交接清单**,不入库。
 
 ## 它是什么
-admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查穿山甲**。穿山甲只通过 GroMore 数据 API 给数、且 **T+1**(次日约 10:00 出昨天的数),所以每天得拉一次入库,报表才会往前走。
+admin「广告收益报表」里的 GroMore / ADN 收益读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查询**。GroMore 的 T+1 初值约 10:00 可用，但第三方 ADN Reporting 数据可能到 13:50 才更新，所以需要早晚各拉一次。
 
-- 每天 10:30 跑一轮 `scripts/sync_pangle_revenue.py`,默认 `--days 3` 回补近 3 天。
+- 每天 10:30 拉初值、14:30 拉日终值，均由 `scripts/sync_pangle_revenue.py` 以 `--days 3` 回补近 3 天。
-- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(预估)+ `api_revenue`(结算口径)。
+- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(排序价预估)+ `api_revenue`(ADN Reporting 回传，更接近结算)。
 - **幂等 upsert**:同一(日期×应用×代码位)重跑只覆盖、不重复,故回补 / 重跑 / catch-up 都安全。
 - 穿山甲无用户/设备维度 → 只能落「汇总/趋势级」,报表带 user_id 过滤时这块收益置空(显示「-」)。
 
@@ -30,15 +30,15 @@ admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**
 ```bash
 sudo cp deploy/pangle-revenue.{service,timer} /etc/systemd/system/
 sudo systemctl daemon-reload && sudo systemctl enable --now pangle-revenue.timer
-systemctl list-timers pangle-revenue.timer   # 确认下次触发时间(应是次日 10:30)
+systemctl list-timers pangle-revenue.timer   # 确认下次触发时间(10:30 或 14:30)
 ```
 
 ## 怎么看健康 / 手动跑一次
 ```bash
-sudo systemctl start pangle-revenue.service     # 立即手动跑一轮(不等 10:30)
+journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 收益合计
-journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 预估收益合计
+sudo systemctl start pangle-revenue.service     # 立即手动跑一轮
 ```
-成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;预估收益合计 ¥19.42`。
+成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;排序价预估合计 ¥19.42`。
 > 看不到收益、提示 `PANGLE_REPORT_* 未配置`→ 回「上线前置」补 `.env`;报 118 → 子账号没授「查看全部数据」。
 
 ## 本机 Windows 开发(无 systemd)
@@ -57,11 +57,11 @@ journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入
 - `--start / --end`:指定闭区间(跨度 ≤ 31 天,接口上限 1 个月,超了报 114)。
 
 ## 注意事项
-- **触发时间**:`OnCalendar=*-*-* 10:30:00`。穿山甲 ~10:00 出数,故别早于 10:00 跑(会拉到空/不全)。
+- **触发时间**:10:30 提供初值，14:30 覆盖为日终值；报表只把 D+1 14:00 后同步的数据标记为日终。
 - **catch-up**:`Persistent=true` 补跑错过的那一轮;叠加 `--days 3`,漏一两天重新触发即自愈。
 - **今天 / 今天以前要分开查**:脚本默认只拉昨天及更早,不混查今天(接口约束),无需关心。
 - **join key 是 `ad_unit_id`(我们配的 104xxx)不是 `code_id`**:`code_id` 是底层各 ADN 代码位,对不上口径;`ad_unit_id='-1'` 是未归因桶。改维度时务必注意(详见脚本头注释)。
-- **`api_revenue` 很稀疏**:测试应用 ADN 没配 Reporting → 全 0,仅 prod 个别位有;`revenue`(预估)才是稳的主力。
+- **`api_revenue` 依赖 ADN Reporting 配置**:未配置的测试应用可能为空或 0；`revenue` 只是排序价估算，不能当结算收入。
 - **DB 无关**:sqlite / postgres 均可(upsert 逐行 select-then-write,不像美团 ETL 需要 PG)。
 - **别和别的触发方式双跑**:本 systemd timer 与「手动 cron / 进程内任务」二选一,虽幂等不会重复入库,纯属多余。
 - **改脚本 / 改部署**:走 git + PR,由有 root 的人部署。

deploy/pangle-revenue.service

@@ -1,5 +1,5 @@
-# 每天拉穿山甲 GroMore T+1 天级收益入库 —— 单轮跑,由 pangle-revenue.timer 每天 10:30 触发。
+# 拉 GroMore T+1 天级收益入库 —— 单轮跑,由 timer 每天 10:30、14:30 触发。
-# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的「穿山甲后台收益(T+1)」区块。
+# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的 GroMore/ADN 对账区块。
 #
 # 仅用于 Linux 服务器;本机 Windows 开发无 systemd,直接手动跑脚本即可:
 #   .venv\Scripts\python -m scripts.sync_pangle_revenue            # 拉昨天(北京时间)

deploy/pangle-revenue.timer

@@ -1,11 +1,12 @@
-# 每天 10:30 触发一次穿山甲 GroMore T+1 收益拉取入库(Linux 服务器用)。
+# 每天 10:30 首次拉取、14:30 终值复拉 GroMore T+1 收益(Linux 服务器用)。
 # 见 pangle-revenue.service 顶部注释的部署步骤。
 [Unit]
-Description=Run Pangle GroMore daily revenue sync at 10:30
+Description=Run Pangle GroMore daily revenue sync at 10:30 and 14:30
 
 [Timer]
-# 穿山甲 T+1、次日约 10:00 出数;10:30 触发留 30min 余量。要错开整点扎堆可微调到 10:35。
+# 10:30 尽早展示初值；第三方 ADN Reporting 最晚约 13:50 更新，14:30 再拉一次作为日终值。
 OnCalendar=*-*-* 10:30:00
+OnCalendar=*-*-* 14:30:00
 # 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日);叠加 --days 3 回补,漏一两天能自愈。
 Persistent=true
 AccuracySec=1min

scripts/seed_admin_revenue_mock.py

@@ -0,0 +1,211 @@
+"""生成本地 admin「广告收益」与数据大盘用的可重复 mock 数据。
+
+只处理 ``local-admin-revenue-mock-`` 前缀的数据，重跑会替换自身数据，不会触碰真实本地记录。
+会覆盖 Draw（含一条历史 feed）、福利激励视频、提现视频，以及对应的金币流水。
+
+用法：
+    python -m scripts.seed_admin_revenue_mock
+"""
+from __future__ import annotations
+
+from datetime import datetime, time, timedelta
+
+from sqlalchemy import delete, select
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ, cn_today
+from app.db.session import SessionLocal
+from app.models.ad_ecpm import AdEcpmRecord
+from app.models.ad_feed_reward import AdFeedRewardRecord
+from app.models.ad_reward import AdRewardRecord
+from app.models.user import User
+from app.models.wallet import CoinAccount, CoinTransaction
+
+PREFIX = "local-admin-revenue-mock-"
+PHONE = "19900009002"
+USERNAME = "80000009002"
+
+
+def _at(day_offset: int, hour: int, minute: int) -> datetime:
+    day = cn_today() - timedelta(days=day_offset)
+    return datetime.combine(day, time(hour, minute), tzinfo=CN_TZ)
+
+
+def _add_coin(
+    db,
+    *,
+    user_id: int,
+    amount: int,
+    biz_type: str,
+    ref_id: str,
+    created_at: datetime,
+    balance_after: int,
+) -> None:
+    db.add(CoinTransaction(
+        user_id=user_id,
+        amount=amount,
+        balance_after=balance_after,
+        biz_type=biz_type,
+        ref_id=ref_id,
+        remark="本地运营后台广告收益 Mock",
+        created_at=created_at,
+    ))
+
+
+def seed() -> dict[str, int]:
+    if settings.APP_ENV == "prod":
+        raise RuntimeError("Refusing to seed admin revenue mock data in production")
+
+    with SessionLocal() as db:
+        # 清理顺序按外键依赖从流水/奖励到展示；只碰本脚本自己的稳定前缀。
+        db.execute(delete(CoinTransaction).where(CoinTransaction.ref_id.like(f"{PREFIX}%")))
+        db.execute(delete(AdFeedRewardRecord).where(
+            AdFeedRewardRecord.client_event_id.like(f"{PREFIX}%")
+        ))
+        db.execute(delete(AdRewardRecord).where(AdRewardRecord.trans_id.like(f"{PREFIX}%")))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.like(f"{PREFIX}%")))
+
+        user = db.execute(select(User).where(User.phone == PHONE)).scalar_one_or_none()
+        if user is None:
+            user = User(
+                phone=PHONE,
+                username=USERNAME,
+                nickname="运营收益 Mock 用户",
+                register_channel="sms",
+                status="active",
+            )
+            db.add(user)
+            db.flush()
+        else:
+            user.nickname = "运营收益 Mock 用户"
+            user.status = "active"
+
+        balance = 0
+        event_count = 0
+        reward_count = 0
+        # 近四天的数据既能覆盖单日，也能覆盖近 7 天趋势与分类合计。
+        for day_offset in range(4):
+            suffix = f"d{day_offset}"
+            compare_trace = "local-invite-mock-compare-success"
+            coupon_trace = "mock-coupon-repeat-prod-second"
+
+            draw_events = [
+                ("draw", "comparison", compare_trace, 2800 + day_offset * 100, 10, 10),
+                ("draw", "coupon", coupon_trace, 1750 + day_offset * 100, 10, 28),
+            ]
+            # 历史 feed 必须被 Draw 分类一起计算，用于走查兼容逻辑。
+            if day_offset == 1:
+                draw_events.append(("feed", "coupon", coupon_trace, 1250, 11, 12))
+
+            for index, (ad_type, scene, trace_id, ecpm, hour, minute) in enumerate(draw_events, start=1):
+                session = f"{PREFIX}{suffix}-draw-{index}"
+                created_at = _at(day_offset, hour, minute)
+                db.add(AdEcpmRecord(
+                    user_id=user.id,
+                    ad_type=ad_type,
+                    feed_scene=scene,
+                    trace_id=trace_id,
+                    ad_session_id=session,
+                    adn="pangle" if index == 1 else "gdt",
+                    slot_id="mock-draw-rit",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw=str(ecpm),
+                    report_date=created_at.date().isoformat(),
+                    created_at=created_at,
+                ))
+                coin = 18 + day_offset * 2
+                db.add(AdFeedRewardRecord(
+                    client_event_id=f"{PREFIX}{suffix}-feed-reward-{index}",
+                    ad_session_id=session,
+                    user_id=user.id,
+                    reward_date=created_at.date().isoformat(),
+                    duration_seconds=20,
+                    unit_count=2,
+                    ecpm_raw=str(ecpm),
+                    adn="pangle" if index == 1 else "gdt",
+                    slot_id="mock-draw-rit",
+                    ad_type=ad_type,
+                    feed_scene=scene,
+                    trace_id=trace_id,
+                    app_env="prod",
+                    our_code_id="104098712",
+                    coin=coin,
+                    status="granted",
+                    created_at=created_at + timedelta(seconds=20),
+                ))
+                balance += coin
+                _add_coin(
+                    db,
+                    user_id=user.id,
+                    amount=coin,
+                    biz_type="feed_ad_reward",
+                    ref_id=f"{PREFIX}{suffix}-feed-coin-{index}",
+                    created_at=created_at + timedelta(seconds=20),
+                    balance_after=balance,
+                )
+                event_count += 1
+                reward_count += 1
+
+            for ad_type, ecpm, hour, coin in (
+                ("reward_video", 13200 + day_offset * 500, 14, 66),
+                ("withdrawal_video", 32000 + day_offset * 800, 18, 0),
+            ):
+                session = f"{PREFIX}{suffix}-{ad_type}"
+                created_at = _at(day_offset, hour, 6)
+                db.add(AdEcpmRecord(
+                    user_id=user.id,
+                    ad_type=ad_type,
+                    ad_session_id=session,
+                    adn="ks" if ad_type == "reward_video" else "baidu",
+                    slot_id="mock-video-rit",
+                    app_env="prod",
+                    our_code_id="104099389",
+                    ecpm_raw=str(ecpm),
+                    report_date=created_at.date().isoformat(),
+                    created_at=created_at,
+                ))
+                event_count += 1
+                if ad_type == "reward_video":
+                    db.add(AdRewardRecord(
+                        trans_id=f"{PREFIX}{suffix}-reward-video",
+                        user_id=user.id,
+                        coin=coin,
+                        status="granted",
+                        reward_scene="reward_video",
+                        ad_session_id=session,
+                        ecpm_raw=str(ecpm),
+                        app_env="prod",
+                        our_code_id="104099389",
+                        reward_date=created_at.date().isoformat(),
+                        reward_name="Mock 福利视频",
+                        created_at=created_at + timedelta(seconds=35),
+                    ))
+                    balance += coin
+                    _add_coin(
+                        db,
+                        user_id=user.id,
+                        amount=coin,
+                        biz_type="reward_video",
+                        ref_id=f"{PREFIX}{suffix}-reward-video-coin",
+                        created_at=created_at + timedelta(seconds=35),
+                        balance_after=balance,
+                    )
+                    reward_count += 1
+
+        account = db.get(CoinAccount, user.id)
+        if account is None:
+            account = CoinAccount(user_id=user.id)
+            db.add(account)
+        account.coin_balance = balance
+        account.total_coin_earned = balance
+        db.commit()
+        return {"events": event_count, "rewards": reward_count, "coin": balance}
+
+
+if __name__ == "__main__":
+    result = seed()
+    print(
+        "Seeded local admin revenue mock: "
+        f"{result['events']} impressions, {result['rewards']} rewards, {result['coin']} coins"
+    )

scripts/sync_pangle_revenue.py

@@ -1,7 +1,7 @@
-"""每日拉取穿山甲 GroMore 天级收益报表入库(供 admin 广告收益报表的「穿山甲后台收益」)。
+"""每日拉取 GroMore 排序价预估与 ADN Reporting 收益入库(供 admin 广告收益对账)。
 
-GroMore 数据 API 为 T+1:次日穿山甲约 10:00 出数。建议线上每天 ~10:30 由 systemd timer 跑一次
+GroMore 数据 API 为 T+1:次日约 10:00 出初值，第三方 ADN Reporting 最晚约 13:50 更新。
-(默认拉【昨天】);穿山甲对历史数据可能订正,故支持回补近 N 天(幂等 upsert,重跑无害)。
+线上由 systemd timer 在 10:30、14:30 各跑一次;历史数据可能订正,故支持回补近 N 天。
 
 用法:
     python -m scripts.sync_pangle_revenue                 # 拉昨天(北京时间)
@@ -117,7 +117,7 @@ def sync_range(start_date: str, end_date: str) -> None:
         stats = repo.upsert_daily_rows(db, rows)
     total_rev = round(sum(r["revenue_yuan"] for r in rows), 4)
     print(f"✅ 完成:接口 {len(raw)} 行 → 入库 {len(rows)} 行(跳过 {skipped}),"
-          f"新增 {stats['inserted']} / 更新 {stats['updated']};预估收益合计 ¥{total_rev}")
+          f"新增 {stats['inserted']} / 更新 {stats['updated']};排序价预估合计 ¥{total_rev}")
 
 
 def main() -> None:

tests/test_admin_ad_revenue_scope.py

@@ -15,6 +15,8 @@ from app.models.user import User
 
 REPORT_DATE = "2040-02-03"
 PLAYBACK_DATE = "2040-02-04"
+DETAIL_DATE = "2040-02-06"
+SOURCE_FALLBACK_DATE = "2040-02-07"
 
 
 def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -> None:
@@ -51,18 +53,22 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="prod", our_code_id="prod-reward",
                 adn="", revenue_yuan=1.5, api_revenue_yuan=1.2, impressions=10,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="prod", our_code_id="prod-demo",
                 adn="", revenue_yuan=8.0, api_revenue_yuan=7.0, impressions=40,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="prod", our_code_id="104098712",
                 adn="", revenue_yuan=2.5, api_revenue_yuan=2.0, impressions=20,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="test", our_code_id="104127529",
                 adn="", revenue_yuan=9.0, api_revenue_yuan=8.0, impressions=50,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
         ])
         db.commit()
@@ -87,6 +93,8 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
         assert business["total_revenue_yuan"] == 0.5
         assert business["total_pangle_revenue_yuan"] == 4.0
         assert business["total_pangle_api_revenue_yuan"] == 3.2
+        assert business["pangle_api_revenue_complete"] is True
+        assert business["pangle_latest_synced_at"] is not None
 
         all_codes = ad_revenue.ad_revenue_report(
             db,
@@ -132,7 +140,7 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
         db.close()
 
 
-def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
+def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap() -> None:
     db = SessionLocal()
     phone = "18800009992"
     sessions = {
@@ -148,13 +156,17 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
 
         for index, (status, session_id) in enumerate(sessions.items(), start=1):
             created_at = datetime(2040, 2, 4, index, tzinfo=UTC)
+            # 发奖状态 capped 的广告故意使用 ¥1000 CPM，验证收入不套用金币侧 ¥500 CPM 封顶。
+            ecpm_raw = "100000" if status == "capped" else "10000"
             db.add(AdEcpmRecord(
                 user_id=user.id,
                 ad_type="reward_video",
                 ad_session_id=session_id,
+                adn=f"adn-{status}",
+                slot_id=f"rit-{status}",
                 app_env="prod",
                 our_code_id="prod-reward",
-                ecpm_raw="10000",
+                ecpm_raw=ecpm_raw,
                 report_date=PLAYBACK_DATE,
                 created_at=created_at,
             ))
@@ -167,7 +179,7 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
                 ad_session_id=session_id,
                 app_env="prod",
                 our_code_id="prod-reward",
-                ecpm_raw="10000",
+                ecpm_raw=ecpm_raw,
                 reward_date=PLAYBACK_DATE,
                 created_at=created_at,
             ))
@@ -186,22 +198,33 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
 
         revenue_by_status = {row["status"]: row["revenue_yuan"] for row in result["items"]}
         assert revenue_by_status == {
-            "closed_early": 0.0,
+            "closed_early": 0.1,
-            "too_short": 0.0,
+            "too_short": 0.1,
-            "capped": 0.1,
+            "capped": 1.0,
             "granted": 0.1,
         }
         assert result["total_impressions"] == 4
-        assert result["total_revenue_yuan"] == 0.2
+        assert result["total_revenue_yuan"] == 1.3
         assert len(result["daily"]) == 1
         assert result["daily"][0]["date"] == PLAYBACK_DATE
         assert result["daily"][0]["impressions"] == 4
-        assert result["daily"][0]["revenue_yuan"] == 0.2
+        assert result["daily"][0]["revenue_yuan"] == 1.3
-        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 0.2
+        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 1.3
         assert result["type_stats"]["reward_video"] == {
             "impressions": 4,
-            "revenue_yuan": 0.2,
+            "revenue_yuan": 1.3,
+            "ecpm_yuan": 325.0,
         }
+        detail_by_status = {
+            row["status"]: row["reward_detail"] for row in result["items"]
+        }
+        assert detail_by_status["granted"]["adn"] == "adn-granted"
+        assert detail_by_status["granted"]["slot_id"] == "rit-granted"
+        # 未进入发奖的记录可保留展示收入，但不能凭空生成奖励因子或占用 LT 累计。
+        for status in ("closed_early", "too_short", "capped"):
+            assert detail_by_status[status]["ecpm_factor"] is None
+            assert detail_by_status[status]["lt_factor_start"] is None
+            assert detail_by_status[status]["lt_index_start"] is None
     finally:
         db.rollback()
         db.execute(delete(AdRewardRecord).where(AdRewardRecord.reward_date == PLAYBACK_DATE))
@@ -211,54 +234,133 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         db.close()
 
 
-def test_feed_reward_details_keep_each_record_adn() -> None:
+def test_category_stats_merge_legacy_feed_and_withdrawal_video() -> None:
+    db = SessionLocal()
+    phone = "18800009993"
+    category_date = "2040-02-05"
+    try:
+        user = User(phone=phone, username="29999999993", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            # Draw 经营分类必须包含新 draw 与历史 feed。
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="category-draw",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="feed", ad_session_id="category-feed",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
+            ),
+            # 看视频经营分类必须包含福利与提现两个视频入口。
+            AdEcpmRecord(
+                user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=category_date,
+            date_to=category_date,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+
+        assert result["category_stats"] == {
+            "draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
+            "video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
+def test_feed_reward_detail_keeps_each_record_adn_instead_of_parent_adn() -> None:
     db = SessionLocal()
     phone = "18800009994"
-    detail_date = "2040-02-06"
     try:
         user = User(phone=phone, username="29999999994", register_channel="sms")
         db.add(user)
         db.flush()
         db.add_all([
             AdFeedRewardRecord(
-                client_event_id="detail-adn-pangle", user_id=user.id,
+                client_event_id="detail-adn-pangle",
-                reward_date=detail_date, duration_seconds=20, unit_count=1,
+                user_id=user.id,
-                ecpm_raw="12000", adn="pangle", slot_id="rit-pangle",
+                reward_date=DETAIL_DATE,
-                ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
+                duration_seconds=20,
-                app_env="prod", our_code_id="104098712", coin=12, status="granted",
+                unit_count=1,
+                ecpm_raw="12000",
+                adn="pangle",
+                slot_id="rit-pangle",
+                ad_type="draw",
+                feed_scene="coupon",
+                trace_id="detail-adn-trace",
+                app_env="prod",
+                our_code_id="104098712",
+                coin=12,
+                status="granted",
                 created_at=datetime(2040, 2, 6, 1, tzinfo=UTC),
             ),
             AdFeedRewardRecord(
-                client_event_id="detail-adn-gdt", user_id=user.id,
+                client_event_id="detail-adn-gdt",
-                reward_date=detail_date, duration_seconds=20, unit_count=1,
+                user_id=user.id,
-                ecpm_raw="25000", adn="gdt", slot_id="rit-gdt",
+                reward_date=DETAIL_DATE,
-                ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
+                duration_seconds=20,
-                app_env="prod", our_code_id="104098712", coin=25, status="granted",
+                unit_count=1,
+                ecpm_raw="25000",
+                adn="gdt",
+                slot_id="rit-gdt",
+                ad_type="draw",
+                feed_scene="coupon",
+                trace_id="detail-adn-trace",
+                app_env="prod",
+                our_code_id="104098712",
+                coin=25,
+                status="granted",
                 created_at=datetime(2040, 2, 6, 2, tzinfo=UTC),
             ),
         ])
         db.commit()
 
         result = ad_revenue.ad_revenue_report(
-            db, date_from=detail_date, date_to=detail_date,
+            db,
-            user_id=user.id, app_env="prod", revenue_scope="all",
+            date_from=DETAIL_DATE,
+            date_to=DETAIL_DATE,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
         )
 
         item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
+        assert item["adn"] is None
         assert [detail["adn"] for detail in item["sub_rewards"]] == ["pangle", "gdt"]
         assert [detail["slot_id"] for detail in item["sub_rewards"]] == ["rit-pangle", "rit-gdt"]
     finally:
         db.rollback()
-        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == detail_date))
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == DETAIL_DATE))
         db.execute(delete(User).where(User.phone == phone))
         db.commit()
         db.close()
 
 
-def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
+def test_feed_reward_source_can_fallback_to_unique_trace_impression_only() -> None:
+    """发奖会话是整场 ID、展示会话是 impressionId 时，只在 trace+eCPM 唯一时回填 ADN。"""
     db = SessionLocal()
     phone = "18800009995"
-    fallback_date = "2040-02-07"
     try:
         user = User(phone=phone, username="29999999995", register_channel="sms")
         db.add(user)
@@ -266,7 +368,7 @@ def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
         db.add_all([
             AdFeedRewardRecord(
                 client_event_id="source-fallback-unique", user_id=user.id,
-                reward_date=fallback_date, duration_seconds=3, unit_count=0,
+                reward_date=SOURCE_FALLBACK_DATE, duration_seconds=3, unit_count=0,
                 ecpm_raw="4700", ad_session_id="flow-session", trace_id="source-trace",
                 app_env="prod", our_code_id="104098712", coin=0, status="too_short",
                 ad_type="draw", feed_scene="comparison",
@@ -274,7 +376,7 @@ def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
             ),
             AdFeedRewardRecord(
                 client_event_id="source-fallback-ambiguous", user_id=user.id,
-                reward_date=fallback_date, duration_seconds=3, unit_count=0,
+                reward_date=SOURCE_FALLBACK_DATE, duration_seconds=3, unit_count=0,
                 ecpm_raw="4800", ad_session_id="flow-session", trace_id="source-trace",
                 app_env="prod", our_code_id="104098712", coin=0, status="too_short",
                 ad_type="draw", feed_scene="comparison",
@@ -283,27 +385,31 @@ def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
             AdEcpmRecord(
                 user_id=user.id, ad_type="draw", ad_session_id="impression-unique",
                 trace_id="source-trace", ecpm_raw="4700", adn="baidu", slot_id="rit-baidu",
-                app_env="prod", our_code_id="104098712", report_date=fallback_date,
+                app_env="prod", our_code_id="104098712", report_date=SOURCE_FALLBACK_DATE,
                 created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
             ),
             AdEcpmRecord(
                 user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-a",
                 trace_id="source-trace", ecpm_raw="4800", adn="baidu", slot_id="rit-baidu",
-                app_env="prod", our_code_id="104098712", report_date=fallback_date,
+                app_env="prod", our_code_id="104098712", report_date=SOURCE_FALLBACK_DATE,
                 created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
             ),
             AdEcpmRecord(
                 user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-b",
                 trace_id="source-trace", ecpm_raw="4800", adn="ks", slot_id="rit-ks",
-                app_env="prod", our_code_id="104098712", report_date=fallback_date,
+                app_env="prod", our_code_id="104098712", report_date=SOURCE_FALLBACK_DATE,
                 created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
             ),
         ])
         db.commit()
 
         result = ad_revenue.ad_revenue_report(
-            db, date_from=fallback_date, date_to=fallback_date,
+            db,
-            user_id=user.id, app_env="prod", revenue_scope="all",
+            date_from=SOURCE_FALLBACK_DATE,
+            date_to=SOURCE_FALLBACK_DATE,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
         )
         item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
         details = {detail["ecpm"]: detail for detail in item["sub_rewards"]}
@@ -313,8 +419,27 @@ def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
         assert details["4800"]["slot_id"] is None
     finally:
         db.rollback()
-        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == fallback_date))
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == SOURCE_FALLBACK_DATE))
-        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == fallback_date))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == SOURCE_FALLBACK_DATE))
         db.execute(delete(User).where(User.phone == phone))
         db.commit()
         db.close()
+
+
+def test_pangle_api_day_is_provisional_before_14_beijing_time() -> None:
+    assert ad_revenue._pangle_api_day_complete(
+        REPORT_DATE,
+        {
+            "api_revenue_yuan": 3.2,
+            # D+1 10:30 北京时间。
+            "synced_at": datetime(2040, 2, 4, 2, 30, tzinfo=UTC),
+        },
+    ) is False
+    assert ad_revenue._pangle_api_day_complete(
+        REPORT_DATE,
+        {
+            "api_revenue_yuan": 3.2,
+            # D+1 14:00 北京时间，达到日终判定线。
+            "synced_at": datetime(2040, 2, 4, 6, 0, tzinfo=UTC),
+        },
+    ) is True

tests/test_admin_read.py

@@ -1,7 +1,7 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import datetime
 
 import pytest
 from fastapi.testclient import TestClient
@@ -9,7 +9,6 @@ from sqlalchemy import event
 
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
-from app.admin.repositories import queries
 from app.db.session import SessionLocal, engine
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
@@ -145,8 +144,6 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     ) -> None:
         if "ad_reward_record.boost_round_id" in statement:
             raise AssertionError("提现详情不应查询未使用的 boost_round_id")
-        if "FROM user" in statement and "user.phone" in statement:
-            raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表")
 
     event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
     try:
@@ -165,17 +162,6 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     assert records.status_code == 200, records.text
 
 
-def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
-    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
-    naive = datetime(2038, 1, 1, 8, 0)
-    aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)
-
-    rows = [{"created_at": aware}, {"created_at": naive}]
-    rows.sort(key=queries._coin_record_sort_key, reverse=True)
-
-    assert rows == [{"created_at": naive}, {"created_at": aware}]
-
-
 def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
     _seed_user_with_data("13800000003")
     r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))

tests/test_derive_best_excludes_short.py

@@ -1,102 +0,0 @@
-"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"。
-
-回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
-会把缺菜店的虚低价当 best_price → 记录页戴"最低"红框 + 虚假省额。
-修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)。
-纯函数, 不碰 DB。
-"""
-from app.repositories.comparison import _derive, _derive_from_results
-from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
-
-
-def test_derive_from_results_excludes_short_ordered_from_best():
-    # jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
-    results = [
-        {"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
-        {"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
-        {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
-    ]
-    platform_results = {
-        "jd_waimai": {"skipped_dish_count": 2},
-        "taobao_flash": {"skipped_dish_count": 0},
-        "meituan": {"is_source": True},
-    }
-    d = _derive_from_results(results, platform_results)
-    assert d["best_platform_id"] == "taobao_flash"
-    assert d["best_price_cents"] == 3850
-    assert d["saved_amount_cents"] == 4200 - 3850   # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
-    assert d["is_source_best"] is False
-
-
-def test_derive_from_results_all_targets_short_falls_back_to_source():
-    # 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
-    d = _derive_from_results(results, platform_results)
-    assert d["best_platform_id"] == "meituan"
-    assert d["is_source_best"] is True
-    assert d["saved_amount_cents"] == 0
-
-
-def test_derive_from_results_no_platform_results_keeps_old_behavior():
-    # 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    d = _derive_from_results(results)
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500
-
-
-def test_derive_from_results_malformed_platform_results_no_crash():
-    # 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    d = _derive_from_results(results, {"jd_waimai": "oops"})
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500
-
-
-def test_derive_pydantic_excludes_short():
-    # _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
-    payload = ComparisonRecordIn(
-        trace_id="t-short-pyd",
-        source_price=42.0,
-        source_platform_id="meituan",
-        comparison_results=[
-            ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
-            ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
-            ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
-        ],
-        platform_results={
-            "jd_waimai": {"skipped_dish_count": 2},
-            "taobao_flash": {"skipped_dish_count": 0},
-        },
-    )
-    d = _derive(payload)
-    assert d["best_platform_id"] == "taobao_flash"
-    assert d["best_price_cents"] == 3850
-    assert d["saved_amount_cents"] == 4200 - 3850
-    assert d["is_source_best"] is False
-
-
-def test_derive_pydantic_malformed_platform_results_no_crash():
-    # _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
-    payload = ComparisonRecordIn(
-        trace_id="t-malformed-pyd",
-        source_price=42.0,
-        comparison_results=[
-            ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
-            ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
-        ],
-        platform_results={"jd_waimai": "oops"},
-    )
-    d = _derive(payload)  # 不抛 AttributeError
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500

tests/test_guide_video.py

@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
     assert first["should_play"] is True and first["seq"] == 1
 
     # 本次请求读到的是过期计数 → 仍会算出 seq=1
-    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
+    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
     with SessionLocal() as db:
         result = crud_guide.start_play(db, uid)