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

.env.example

@@ -139,11 +139,6 @@ PRICEBOT_COMPARE_TIMEOUT_SEC=60
 # 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
 # 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
 INTERNAL_API_SECRET=
-# 新版比价 harvest 完成后即时回填；以下 worker 再补偿短暂故障期间漏掉的记录。
-LLM_COST_BACKFILL_ENABLED=true
-LLM_COST_BACKFILL_INTERVAL_SEC=300
-LLM_COST_BACKFILL_BATCH_SIZE=100
-LLM_COST_BACKFILL_LOOKBACK_DAYS=30
 
 # ===== CORS =====
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类

alembic/versions/user_manual_risk_fields.py

@@ -1,76 +0,0 @@
-"""add manual high-risk flag and note to user
-
-Revision ID: user_manual_risk_fields
-Revises: risk_monitor_generic
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "user_manual_risk_fields"
-down_revision: str | None = "risk_monitor_generic"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-_INVITE_WITHDRAW_PAGE = "invite-withdraws"
-
-
-def _finance_role() -> sa.TableClause:
-    return sa.table(
-        "admin_role",
-        sa.column("name", sa.String),
-        sa.column("pages", _JSON),
-    )
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("user") as batch_op:
-        batch_op.add_column(
-            sa.Column(
-                "is_high_risk",
-                sa.Boolean(),
-                server_default=sa.false(),
-                nullable=False,
-            )
-        )
-        batch_op.add_column(sa.Column("high_risk_note", sa.Text(), nullable=True))
-        batch_op.create_index("ix_user_is_high_risk", ["is_high_risk"], unique=False)
-    role = _finance_role()
-    conn = op.get_bind()
-    row = conn.execute(
-        sa.select(role.c.pages).where(role.c.name == "finance")
-    ).scalar_one_or_none()
-    if row is not None and _INVITE_WITHDRAW_PAGE not in (row or []):
-        conn.execute(
-            role.update()
-            .where(role.c.name == "finance")
-            .values(pages=[*(row or []), _INVITE_WITHDRAW_PAGE])
-        )
-
-
-def downgrade() -> None:
-    role = _finance_role()
-    conn = op.get_bind()
-    row = conn.execute(
-        sa.select(role.c.pages).where(role.c.name == "finance")
-    ).scalar_one_or_none()
-    if row is not None:
-        conn.execute(
-            role.update()
-            .where(role.c.name == "finance")
-            .values(
-                pages=[
-                    page for page in (row or []) if page != _INVITE_WITHDRAW_PAGE
-                ]
-            )
-        )
-    with op.batch_alter_table("user") as batch_op:
-        batch_op.drop_index("ix_user_is_high_risk")
-        batch_op.drop_column("high_risk_note")
-        batch_op.drop_column("is_high_risk")

app/admin/permissions.py

@@ -23,8 +23,7 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "cps", "label": "CPS收益"},
     ]},
     {"group": "奖励审核", "pages": [
-        {"key": "invite-withdraws", "label": "邀请提现审核"},
-        {"key": "withdraws", "label": "其他提现审核"},
+        {"key": "withdraws", "label": "提现审核"},
         {"key": "price-reports", "label": "低价审核"},
         {"key": "feedbacks", "label": "用户反馈"},
     ]},
@@ -60,7 +59,7 @@ BUILTIN_ROLES: list[dict] = [
         "cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
     ]},
     {"name": "finance", "label": "财务", "pages": [
-        "dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws",
+        "dashboard", "ad-revenue-report", "cps", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
         "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",

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 的展示记录回填。
-    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
+    # S2S 发奖回调本身不携带实际填充的 ADN/底层 rit；按客户端在展示时上报的
+    # 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
-        for record in db.execute(
+        (rec.user_id, rec.ad_session_id): rec
+        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[
-    dict[tuple[int, str], tuple[str | None, str | None]],
-    dict[tuple[int, str, str], tuple[str | None, str | None]],
-]:
-    """为旧信息流发奖记录构建安全来源索引。"""
+) -> tuple[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
-# 上报 eCPM，随后才在关闭时补报以下终态，因此必须在展示/发奖合并后修正收益。
-_ZERO_REVENUE_REWARD_VIDEO_STATUSES = frozenset({"closed_early", "too_short"})
+# GroMore 官方说明第三方 ADN 的 Reporting API 最晚约 13:50 更新。只有 D+1 14:00
+# 之后完成的同步才标记为「API 同步窗口完成」；这不代表覆盖全部 ADN 或最终结算。
+_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}
-    # 聚合父行可能包含多个 ADN，来源必须保留在每一条发奖明细上。
+    detail = {k: row[k] for k in _REWARD_DETAIL_KEYS}
+    # 发奖明细必须保留自己的广告网络，不能复用整场聚合父行的来源：
+    # 同一次比价/领券可能先后由不同 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)
-            # 再折收益,与发奖口径 [rewards.calculate_ad_reward_coin] 一致(2026-06-29 修:原裸 parse_ecpm_yuan
-            # 不钳,伪造/异常天价 eCPM 会把报表预估收益冲到任意大;金币侧已钳、收益侧漏钳)。
-            "revenue_yuan": round(
-                min(rewards.parse_ecpm_yuan(rec.ecpm_raw), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0, 6,
-            ),
+            # 客户端 SDK 展示预估收益(元)= 后端留存 getEcpm 元/千次 ÷ 1000。
+            # 这里不能复用发奖防作弊的 ¥500 CPM 钳顶：钳顶只限制金币成本，不改变广告已产生的
+            # 收入估值。onAdShow 已发生即计展示收入，是否看满只影响发奖，不影响广告收入。
+            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 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,73 +438,50 @@ def ad_revenue_report(
             for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
         ]
 
-    # 分广告类型小计(按 ad_type:展示条数 + 预估收益;eCPM 由前端用 收益÷展示×1000 算)。
-    # 基于全量(已按 feed_scene 过滤)events;前端只取 draw / reward_video 两类展示。
-    type_map: dict[str, dict] = {}
-    for e in events:
-        t = type_map.get(e["ad_type"])
-        if t is None:
-            t = {"impressions": 0, "revenue_yuan": 0.0}
-            type_map[e["ad_type"]] = t
-        t["impressions"] += e["impressions"]
-        t["revenue_yuan"] += e["revenue_yuan"]
-    type_stats = {
-        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
-        for k, v in type_map.items()
-    }
+    def _aggregate_stats(bucket_of) -> dict[str, dict]:
+        """按展示事件聚合收益 / 加权 SDK eCPM，避免前端漏合并历史类型。"""
+        stat_map: dict[str, dict] = {}
+        for e in events:
+            bucket = bucket_of(e)
+            if bucket is None:
+                continue
+            stat = stat_map.setdefault(bucket, {
+                "impressions": 0,
+                "revenue_yuan": 0.0,
+                "ecpm_fen_sum": 0.0,
+            })
+            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()
+        }
 
-    # 经营看板的两类 eCPM 必须按真实展示的 SDK eCPM 加权，不能用发奖状态修正后的
-    # 收益反推。Draw 包含新 draw 与历史 feed；看视频包含福利与提现视频。
-    category_map: dict[str, dict] = {}
-    for event in events:
-        category = (
-            "draw" if event["ad_type"] in {"draw", "feed"}
-            else "video" if event["ad_type"] in {"reward_video", "withdrawal_video"}
+    # 原始 ad_type 小计，供明细筛选和排查使用。
+    type_stats = _aggregate_stats(lambda e: e["ad_type"])
+    # 经营看板使用的规范分类：Draw 包含历史 feed；看视频包含福利与提现视频。
+    # 这两个集合与筛选逻辑保持一致，避免只取 draw / reward_video 而漏算历史或提现数据。
+    category_stats = _aggregate_stats(
+        lambda e: (
+            "draw" if e["ad_type"] in {"draw", "feed"}
+            else "video" if e["ad_type"] in {"reward_video", "withdrawal_video"}
             else None
         )
-        if category is None:
-            continue
-        stat = category_map.setdefault(category, {
-            "impressions": 0,
-            "revenue_yuan": 0.0,
-            "ecpm_fen_sum": 0.0,
-        })
-        impressions = int(event["impressions"])
-        stat["impressions"] += impressions
-        stat["revenue_yuan"] += event["revenue_yuan"]
-        stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(event["ecpm"]) * impressions
-    category_stats = {
-        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 category_map.items()
-    }
+    )
 
-    # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
-    # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
-    # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
-    # 改为服务端在全量上聚合下发(也顺带不受 limit 分页截断影响)。feed_scene 为空(激励视频 /
-    # 旧数据)不计入任何场景桶。
-    scene_map: dict[str, dict] = {}
-    for e in events:
-        sc = e.get("feed_scene")
-        if not sc:
-            continue
-        s = scene_map.get(sc)
-        if s is None:
-            s = {"impressions": 0, "revenue_yuan": 0.0}
-            scene_map[sc] = s
-        s["impressions"] += e["impressions"]
-        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()
-    }
+    # 分场景小计，同 type_stats 基于全量 events，供数据大盘「领券广告 / 比价广告」卡使用。
+    # feed_scene 为空的激励视频 / 历史数据不计入任何场景桶。
+    scene_stats = _aggregate_stats(lambda e: e.get("feed_scene"))
 
     # DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
     # 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
@@ -507,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,

app/admin/repositories/mutations.py

@@ -44,25 +44,6 @@ def set_user_debug_trace(
     return user
 
 
-def set_user_risk(
-    db: Session,
-    user: User,
-    *,
-    is_high_risk: bool,
-    note: str | None,
-    commit: bool = True,
-) -> User:
-    """设置人工风险结论和备注，支持与审计日志共用同一事务。"""
-    user.is_high_risk = is_high_risk
-    user.high_risk_note = note.strip() if is_high_risk and note else None
-    if commit:
-        db.commit()
-        db.refresh(user)
-    else:
-        db.flush()
-    return user
-
-
 def update_feedback_status(
     db: Session, feedback: Feedback, *, status: str, commit: bool = True
 ) -> Feedback:

app/admin/repositories/queries.py

@@ -9,7 +9,7 @@ from datetime import date, datetime, time, timedelta, timezone
 from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo
 
-from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
+from sqlalchemy import Select, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
@@ -22,15 +22,14 @@ from app.models.comparison import ComparisonRecord
 from app.models.coupon_state import CouponPromptEngagement
 from app.models.device import DeviceLiveness
 from app.models.feedback import Feedback
-from app.models.invite import InviteRelation
 from app.models.onboarding import OnboardingCompletion
 from app.models.price_report import PriceReport
-from app.models.savings import SavingsRecord
 from app.models.user import User
 from app.models.wallet import (
     CashTransaction,
     CoinAccount,
     CoinTransaction,
+    InviteCashTransaction,
     WithdrawOrder,
 )
 from app.repositories import activity, ad_ecpm
@@ -46,78 +45,6 @@ _FEED_SCENE_LABEL = {
 }
 
 
-_DEVICE_MARKETING_NAMES = {
-    "23078RKD5C": "Redmi K60 至尊版",
-    "M2012K11AC": "Redmi K40",
-    "PJA110": "一加 Ace 2 Pro",
-    "PPG-AN00": "荣耀 GT Pro",
-    "V2166BA": "vivo Y77e",
-    "V2309A": "vivo X100",
-}
-
-
-def _device_marketing_name(model: str | None) -> str | None:
-    """把线上已知 Build.MODEL 编码转成用户可识别的商品名。"""
-    if not model:
-        return None
-    return _DEVICE_MARKETING_NAMES.get(model.strip().upper())
-
-
-def _attach_comparison_device_details(items: list[ComparisonRecord]) -> None:
-    """给比价记录补充可读机型名，同时保留原始设备编码。"""
-    for item in items:
-        item.device_model_name = _device_marketing_name(item.device_model)
-
-
-def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
-    """按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
-    candidates = [
-        item for item in feedbacks if item.device_model and item.device_model.strip()
-    ]
-    for item in candidates:
-        item.device_model_name = _device_marketing_name(item.device_model)
-        item.device_manufacturer = None
-        item.rom_version = None
-    if not candidates:
-        return
-
-    ranked = (
-        select(
-            Feedback.id.label("feedback_id"),
-            ComparisonRecord.device_manufacturer.label("device_manufacturer"),
-            ComparisonRecord.rom_version.label("rom_version"),
-            func.row_number()
-            .over(
-                partition_by=Feedback.id,
-                order_by=(
-                    ComparisonRecord.created_at.desc(),
-                    ComparisonRecord.id.desc(),
-                ),
-            )
-            .label("row_num"),
-        )
-        .join(
-            ComparisonRecord,
-            and_(
-                ComparisonRecord.user_id == Feedback.user_id,
-                ComparisonRecord.device_model == Feedback.device_model,
-                ComparisonRecord.created_at <= Feedback.created_at,
-            ),
-        )
-        .where(Feedback.id.in_([item.id for item in candidates]))
-        .subquery()
-    )
-    details = {
-        row.feedback_id: row
-        for row in db.execute(select(ranked).where(ranked.c.row_num == 1)).all()
-    }
-
-    for item in candidates:
-        detail = details.get(item.id)
-        item.device_manufacturer = detail.device_manufacturer if detail else None
-        item.rom_version = detail.rom_version if detail else None
-
-
 def cursor_paginate(
     db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
 ) -> tuple[list, int | None]:
@@ -350,7 +277,6 @@ def list_comparison_records(
         limit=limit, cursor=cursor,
     )
     _attach_user_info(db, items)
-    _attach_comparison_device_details(items)
     # 「本次比价看广告的预估收益」:按本页 trace_id 一次性聚合(同 _attach_user_info 逐页范式)。
     # ad_revenue_yuan 非 ORM 列,仅瞬态挂实例上供 AdminComparisonListItem(from_attributes)读出。
     rev = ad_ecpm.revenue_yuan_by_trace(db, [it.trace_id for it in items])
@@ -497,7 +423,6 @@ def get_comparison_record(db: Session, record_id: int) -> ComparisonRecord | Non
     rec = db.get(ComparisonRecord, record_id)
     if rec is not None:
         _attach_user_info(db, [rec])
-        _attach_comparison_device_details([rec])
     return rec
 
 
@@ -785,7 +710,16 @@ def list_all_withdraw_orders(
     elif quick_filter == "today":
         stmt = stmt.where(WithdrawOrder.created_at >= today_start)
     elif quick_filter == "high_risk":
-        stmt = stmt.where(User.is_high_risk.is_(True))
+        stmt = stmt.where(
+            or_(
+                WithdrawOrder.user_name.is_(None),
+                WithdrawOrder.user_name == "",
+                User.status != "active",
+                User.created_at >= now - timedelta(hours=24),
+                WithdrawOrder.status.in_(("failed", "rejected")),
+                WithdrawOrder.fail_reason.is_not(None),
+            )
+        )
 
     sort_cols = {
         "id": WithdrawOrder.id,
@@ -811,9 +745,7 @@ def _as_utc(value: datetime) -> datetime:
     return value.astimezone(timezone.utc)
 
 
-def withdraw_list_enrichment(
-    db: Session, user_ids: list[int], *, source: str | None = None
-) -> dict[int, dict]:
+def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict]:
     """批量富化提现单列表:按本页 user_id 取 手机号/昵称 + 各自累计成功提现金额(分)。
 
     两条聚合查询搞定(避免逐行 N+1)。累计口径与 get_user_overview 的 withdraw_success_cents
@@ -823,32 +755,25 @@ def withdraw_list_enrichment(
     uniq = list(set(user_ids))
     if not uniq:
         return {}
-    success_stmt = select(
-        WithdrawOrder.user_id,
-        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
-    ).where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
-    if source:
-        success_stmt = success_stmt.where(WithdrawOrder.source == source)
-    success_rows = db.execute(success_stmt.group_by(WithdrawOrder.user_id)).all()
+    success_rows = db.execute(
+        select(
+            WithdrawOrder.user_id,
+            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+        )
+        .where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
+        .group_by(WithdrawOrder.user_id)
+    ).all()
     success_map = {uid: int(total) for uid, total in success_rows}
     users = db.execute(
-        select(
-            User.id,
-            User.phone,
-            User.nickname,
-            User.is_high_risk,
-            User.high_risk_note,
-        ).where(User.id.in_(uniq))
+        select(User.id, User.phone, User.nickname).where(User.id.in_(uniq))
     ).all()
     return {
         uid: {
             "phone": phone,
             "nickname": nickname,
-            "is_high_risk": is_high_risk,
-            "high_risk_note": high_risk_note,
             "cumulative_success_cents": success_map.get(uid, 0),
         }
-        for uid, phone, nickname, is_high_risk, high_risk_note in users
+        for uid, phone, nickname in users
     }
 
 
@@ -892,7 +817,6 @@ def list_feedbacks(
         db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
     )
     _attach_user_info(db, items)  # 列表展示完整手机号(点手机号查该用户全部反馈)
-    _attach_feedback_device_details(db, items)
     return items, next_cursor, total
 
 
@@ -955,16 +879,15 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder
     ).scalar_one_or_none()
 
 
-def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
-    """提现审核台顶部统计。金额单位:分；口径与列表的 source 筛选一致。"""
-    summary_stmt = select(
-        WithdrawOrder.status,
-        func.count(WithdrawOrder.id),
-        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
-    )
-    if source:
-        summary_stmt = summary_stmt.where(WithdrawOrder.source == source)
-    rows = db.execute(summary_stmt.group_by(WithdrawOrder.status)).all()
+def withdraw_summary(db: Session) -> dict:
+    """提现审核台顶部统计。金额单位:分。"""
+    rows = db.execute(
+        select(
+            WithdrawOrder.status,
+            func.count(WithdrawOrder.id),
+            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+        ).group_by(WithdrawOrder.status)
+    ).all()
     by_status = {
         status: {"count": int(count), "amount_cents": int(amount_cents)}
         for status, count, amount_cents in rows
@@ -977,154 +900,31 @@ def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
     )
 
     def _today_count(status: str) -> int:
-        stmt = select(func.count(WithdrawOrder.id)).where(
-            WithdrawOrder.status == status,
+        return db.execute(
+            select(func.count(WithdrawOrder.id)).where(
+                WithdrawOrder.status == status,
+                WithdrawOrder.updated_at >= today_start,
+            )
+        ).scalar_one()
+
+    today_success_amount = db.execute(
+        select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
+            WithdrawOrder.status == "success",
             WithdrawOrder.updated_at >= today_start,
         )
-        if source:
-            stmt = stmt.where(WithdrawOrder.source == source)
-        return db.execute(stmt).scalar_one()
-
-    today_success_amount_stmt = select(
-        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)
-    ).where(
-        WithdrawOrder.status == "success",
-        WithdrawOrder.updated_at >= today_start,
-    )
-    if source:
-        today_success_amount_stmt = today_success_amount_stmt.where(
-            WithdrawOrder.source == source
-        )
-    today_success_amount = db.execute(today_success_amount_stmt).scalar_one()
-
-    def _status_count(status: str) -> int:
-        return by_status.get(status, {}).get("count", 0)
+    ).scalar_one()
 
     return {
-        "reviewing_count": _status_count("reviewing"),
+        "reviewing_count": by_status.get("reviewing", {}).get("count", 0),
         "reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
-        "pending_count": _status_count("pending"),
-        "success_count": _status_count("success"),
-        "rejected_count": _status_count("rejected"),
-        "failed_count": _status_count("failed"),
-        "total_count": sum(item["count"] for item in by_status.values()),
+        "pending_count": by_status.get("pending", {}).get("count", 0),
+        "failed_count": by_status.get("failed", {}).get("count", 0),
         "today_success_count": _today_count("success"),
         "today_success_amount_cents": int(today_success_amount),
         "today_rejected_count": _today_count("rejected"),
     }
 
 
-def invite_overview(
-    db: Session,
-    inviter_user_id: int,
-    *,
-    date_from: datetime | None = None,
-    date_to: datetime | None = None,
-) -> dict:
-    """邀请提现详情：邀请关系统计及受邀用户首次比价/首单信息。
-
-    首次记录按自增 id 取最早一条，批量查询避免按受邀用户逐行查询。
-    邀请成功口径为完成首次比价并已发邀请奖励(compare_reward_granted)。
-    """
-    relation_stmt = (
-        select(InviteRelation, User)
-        .join(User, User.id == InviteRelation.invitee_user_id)
-        .where(InviteRelation.inviter_user_id == inviter_user_id)
-        .order_by(InviteRelation.created_at.asc(), InviteRelation.id.asc())
-    )
-    if date_from is not None:
-        relation_stmt = relation_stmt.where(User.created_at >= _as_utc_naive(date_from))
-    if date_to is not None:
-        relation_stmt = relation_stmt.where(User.created_at <= _as_utc_naive(date_to))
-    relation_rows = db.execute(relation_stmt).all()
-    invitee_ids = [relation.invitee_user_id for relation, _ in relation_rows]
-    if not invitee_ids:
-        return {"invite_total": 0, "invite_success_total": 0, "items": []}
-
-    first_compare_ids = list(
-        db.execute(
-            select(func.min(ComparisonRecord.id))
-            .where(
-                ComparisonRecord.user_id.in_(invitee_ids),
-                ComparisonRecord.status == "success",
-            )
-            .group_by(ComparisonRecord.user_id)
-        ).scalars()
-    )
-    comparisons = (
-        db.execute(
-            select(
-                ComparisonRecord.user_id,
-                ComparisonRecord.store_name,
-                ComparisonRecord.product_names,
-            ).where(ComparisonRecord.id.in_(first_compare_ids))
-        ).all()
-        if first_compare_ids
-        else []
-    )
-    comparison_map = {
-        user_id: (store_name, product_names)
-        for user_id, store_name, product_names in comparisons
-    }
-
-    first_order_ids = list(
-        db.execute(
-            select(func.min(SavingsRecord.id))
-            .where(
-                SavingsRecord.user_id.in_(invitee_ids),
-                SavingsRecord.source == "compare",
-            )
-            .group_by(SavingsRecord.user_id)
-        ).scalars()
-    )
-    orders = (
-        db.execute(
-            select(
-                SavingsRecord.user_id,
-                SavingsRecord.shop_name,
-                SavingsRecord.title,
-                SavingsRecord.dishes,
-                SavingsRecord.order_amount_cents,
-            ).where(SavingsRecord.id.in_(first_order_ids))
-        ).all()
-        if first_order_ids
-        else []
-    )
-    order_map = {
-        user_id: (
-            shop_name,
-            title or "、".join(dishes or []),
-            order_amount_cents,
-        )
-        for user_id, shop_name, title, dishes, order_amount_cents in orders
-    }
-
-    items = []
-    for relation, user in relation_rows:
-        compare_store, compare_products = comparison_map.get(user.id, (None, None))
-        order_store, order_products, order_amount = order_map.get(
-            user.id, (None, None, None)
-        )
-        items.append(
-            {
-                "user_id": user.id,
-                "phone": user.phone,
-                "registered_at": user.created_at,
-                "invite_success": bool(relation.compare_reward_granted),
-                "first_compare_store": compare_store,
-                "first_compare_products": compare_products,
-                "first_order_store": order_store,
-                "first_order_products": order_products,
-                "first_order_amount_cents": order_amount,
-            }
-        )
-    return {
-        "invite_total": len(items),
-        "invite_success_total": sum(1 for item in items if item["invite_success"]),
-        "items": items,
-    }
-
-
 def list_withdraw_audit_logs(
     db: Session, out_bill_no: str, *, limit: int = 20
 ) -> list[AdminAuditLog]:
@@ -1145,6 +945,8 @@ def withdraw_risk_flags(
     cash_balance_cents: int,
 ) -> tuple[list[str], int]:
     flags: list[str] = []
+    if not order.user_name:
+        flags.append("缺少提现实名")
     if user and user.status != "active":
         flags.append(f"账号状态:{user.status}")
     if user and user.created_at:
@@ -1167,6 +969,125 @@ def withdraw_risk_flags(
     return flags, score
 
 
+def _check_withdraw_ledger_side(
+    orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str
+) -> dict:
+    """对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。
+
+    orders 已按 source 过滤到本账;txns 是本账流水表里 withdraw_biz/refund_biz 两类流水。
+    规则:每单发起应有一条扣款流水(ref_id=out_bill_no);失败/拒绝单应有且仅一条退款流水;
+    非退款终态不应出现退款流水。四个计数全为 0 即本账自洽。
+    """
+    withdraw_refs = {txn.ref_id for txn in txns if txn.biz_type == withdraw_biz}
+    refund_counts: dict[str, int] = {}
+    for txn in txns:
+        if txn.biz_type == refund_biz and txn.ref_id:
+            refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
+
+    missing_withdraw = 0
+    missing_refund = 0
+    refund_on_non_terminal = 0
+    for order in orders:
+        if order.out_bill_no not in withdraw_refs:
+            missing_withdraw += 1
+        has_refund = refund_counts.get(order.out_bill_no, 0) > 0
+        if order.status in {"failed", "rejected"} and not has_refund:
+            missing_refund += 1
+        if has_refund and order.status not in {"failed", "rejected"}:
+            refund_on_non_terminal += 1
+
+    return {
+        "missing_withdraw": missing_withdraw,
+        "missing_refund": missing_refund,
+        "duplicate_refund": sum(1 for count in refund_counts.values() if count > 1),
+        "refund_on_non_terminal": refund_on_non_terminal,
+    }
+
+
+def withdraw_ledger_check(db: Session) -> dict:
+    """现金账本校验:两本物理隔离的账各自对账(产品红线:coin_cash / invite_cash 不串)。
+
+    普通现金:CoinAccount.cash_balance_cents ↔ cash_transaction(withdraw/withdraw_refund);
+    邀请奖励金:CoinAccount.invite_cash_balance_cents ↔ invite_cash_transaction
+              (invite_withdraw/invite_withdraw_refund)。
+    提现单按 source 分流到对应账核对——邀请提现的流水写在 invite_cash_transaction 表,
+    绝不能拿去和普通现金流水比(否则每笔邀请提现单都会被误报「缺扣款/缺退款流水」)。
+    分流口径与 create_withdraw 一致:仅 source==invite_cash 走邀请账,其余(含历史空值)归普通现金。
+    """
+    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
+    coin_orders = [o for o in orders if o.source != "invite_cash"]
+    invite_orders = [o for o in orders if o.source == "invite_cash"]
+
+    # —— 普通现金账(coin_cash) ——
+    cash_balance_total = int(
+        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
+    )
+    cash_txn_total = int(
+        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
+    )
+    cash_txns = list(
+        db.execute(
+            select(CashTransaction).where(
+                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    coin = _check_withdraw_ledger_side(
+        coin_orders, cash_txns, withdraw_biz="withdraw", refund_biz="withdraw_refund"
+    )
+    cash_diff = cash_balance_total - cash_txn_total
+
+    # —— 邀请奖励金账(invite_cash,独立账户 + 独立流水表) ——
+    invite_balance_total = int(
+        db.execute(
+            select(func.coalesce(func.sum(CoinAccount.invite_cash_balance_cents), 0))
+        ).scalar_one()
+    )
+    invite_txn_total = int(
+        db.execute(
+            select(func.coalesce(func.sum(InviteCashTransaction.amount_cents), 0))
+        ).scalar_one()
+    )
+    invite_txns = list(
+        db.execute(
+            select(InviteCashTransaction).where(
+                InviteCashTransaction.biz_type.in_(("invite_withdraw", "invite_withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    invite = _check_withdraw_ledger_side(
+        invite_orders, invite_txns,
+        withdraw_biz="invite_withdraw", refund_biz="invite_withdraw_refund",
+    )
+    invite_diff = invite_balance_total - invite_txn_total
+
+    ok = (
+        cash_diff == 0
+        and invite_diff == 0
+        and all(v == 0 for v in coin.values())
+        and all(v == 0 for v in invite.values())
+    )
+    return {
+        "ok": ok,
+        # 普通现金账(coin_cash:金币兑换的现金)
+        "cash_balance_total_cents": cash_balance_total,
+        "cash_transaction_total_cents": cash_txn_total,
+        "balance_diff_cents": cash_diff,
+        "missing_withdraw_txn_count": coin["missing_withdraw"],
+        "missing_refund_txn_count": coin["missing_refund"],
+        "duplicate_refund_txn_count": coin["duplicate_refund"],
+        "refund_txn_on_non_terminal_count": coin["refund_on_non_terminal"],
+        # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)
+        "invite_cash_balance_total_cents": invite_balance_total,
+        "invite_cash_transaction_total_cents": invite_txn_total,
+        "invite_balance_diff_cents": invite_diff,
+        "invite_missing_withdraw_txn_count": invite["missing_withdraw"],
+        "invite_missing_refund_txn_count": invite["missing_refund"],
+        "invite_duplicate_refund_txn_count": invite["duplicate_refund"],
+        "invite_refund_txn_on_non_terminal_count": invite["refund_on_non_terminal"],
+    }
+
+
 def get_user_overview(db: Session, user_id: int) -> dict | None:
     """用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
     user = db.get(User, user_id)
@@ -1226,7 +1147,6 @@ def user_reward_stats(
     *,
     date_from: datetime | None = None,
     date_to: datetime | None = None,
-    withdraw_source: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
@@ -1234,35 +1154,22 @@ def user_reward_stats(
     平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
-    withdraw_source_conds = (
-        [WithdrawOrder.source == withdraw_source] if withdraw_source else []
-    )
     wd_success = db.execute(
         select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
             WithdrawOrder.user_id == user_id,
             WithdrawOrder.status == "success",
-            *withdraw_source_conds,
             *_window_conds(WithdrawOrder.created_at, date_from, date_to),
         )
     ).scalar_one()
     wd_total = db.execute(
         select(func.count(WithdrawOrder.id)).where(
             WithdrawOrder.user_id == user_id,
-            *withdraw_source_conds,
             *_window_conds(WithdrawOrder.created_at, date_from, date_to),
         )
     ).scalar_one()
 
     acc = db.get(CoinAccount, user_id)  # 现金余额:当前快照,不随窗口
-    cash_balance = (
-        (
-            acc.invite_cash_balance_cents
-            if withdraw_source == "invite_cash"
-            else acc.cash_balance_cents
-        )
-        if acc
-        else 0
-    )
+    cash_balance = acc.cash_balance_cents if acc else 0
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
@@ -1321,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,
@@ -1417,10 +1319,7 @@ def user_coin_records(
             "coin": rec.amount,
         })
 
-    # SQLite 常返回 naive datetime，PostgreSQL timestamptz 返回 aware datetime；
-    # 统一成 aware UTC 排序，避免线上合并广告记录与签到记录时抛
-    # “can't compare offset-naive and offset-aware datetimes”。
-    rows.sort(key=_coin_record_sort_key, reverse=True)
+    rows.sort(key=lambda r: r["created_at"], 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/users.py

@@ -16,7 +16,6 @@ from app.admin.schemas.user import (
     GrantCashRequest,
     GrantCoinsRequest,
     SetDebugTraceRequest,
-    SetUserRiskRequest,
     SetUserStatusRequest,
     UserCoinRecord,
     UserRewardStats,
@@ -85,21 +84,12 @@ def get_user_reward_stats(
     db: AdminDb,
     date_from: Annotated[datetime | None, Query()] = None,
     date_to: Annotated[datetime | None, Query()] = None,
-    withdraw_source: Annotated[
-        str | None, Query(pattern="^(coin_cash|invite_cash)$")
-    ] = 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,
-            withdraw_source=withdraw_source,
-        )
+        **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)
     )
 
 
@@ -148,50 +138,6 @@ def set_user_status(
     return OkResponse()
 
 
-@router.post("/{user_id}/risk", response_model=OkResponse, summary="设置人工高风险标记与备注")
-def set_user_risk(
-    user_id: int,
-    body: SetUserRiskRequest,
-    request: Request,
-    admin: Annotated[
-        AdminUser, Depends(require_role("operator", "finance"))
-    ],
-    db: AdminDb,
-) -> OkResponse:
-    user = user_repo.get_user_by_id(db, user_id)
-    if user is None:
-        raise HTTPException(status_code=404, detail="用户不存在")
-    before = {
-        "is_high_risk": user.is_high_risk,
-        "high_risk_note": user.high_risk_note,
-    }
-    mutations.set_user_risk(
-        db,
-        user,
-        is_high_risk=body.is_high_risk,
-        note=body.note,
-        commit=False,
-    )
-    write_audit(
-        db,
-        admin,
-        action="user.risk.set",
-        target_type="user",
-        target_id=user_id,
-        detail={
-            "before": before,
-            "after": {
-                "is_high_risk": user.is_high_risk,
-                "high_risk_note": user.high_risk_note,
-            },
-        },
-        ip=get_client_ip(request),
-        commit=False,
-    )
-    db.commit()
-    return OkResponse()
-
-
 @router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限")
 def set_user_debug_trace(
     user_id: int,

app/admin/routers/withdraw.py

@@ -1,4 +1,4 @@
-"""admin 提现:列表(读)+ 单笔/批量查单与审核操作(写,带审计)。
+"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
 
 提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
 这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
@@ -14,16 +14,17 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.repositories import queries
-from app.admin.schemas.admin import AdminAuditLogOut
 from app.admin.schemas.common import CursorPage
+from app.admin.schemas.admin import AdminAuditLogOut
 from app.admin.schemas.wallet import (
     CashTxnOut,
-    InviteOverviewOut,
-    WithdrawBulkItemResult,
+    ReconcileResult,
     WithdrawBulkRejectRequest,
     WithdrawBulkRequest,
     WithdrawBulkResult,
+    WithdrawBulkItemResult,
     WithdrawDetailOut,
+    WithdrawLedgerCheckOut,
     WithdrawListItemOut,
     WithdrawOrderOut,
     WithdrawRejectRequest,
@@ -82,16 +83,8 @@ def list_withdraws(
         cursor=cursor,
     )
     # 联表带出手机号/昵称 + 累计成功提现(本页 user_id 批量富化,2 条聚合查询,无 N+1)。
-    enrichment = queries.withdraw_list_enrichment(
-        db, [o.user_id for o in items], source=source
-    )
-    _empty = {
-        "phone": None,
-        "nickname": None,
-        "is_high_risk": False,
-        "high_risk_note": None,
-        "cumulative_success_cents": 0,
-    }
+    enrichment = queries.withdraw_list_enrichment(db, [o.user_id for o in items])
+    _empty = {"phone": None, "nickname": None, "cumulative_success_cents": 0}
     out_items = [
         WithdrawListItemOut.model_validate(o).model_copy(
             update=enrichment.get(o.user_id, _empty)
@@ -102,13 +95,8 @@ def list_withdraws(
 
 
 @router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
-def withdraws_summary(
-    db: AdminDb,
-    source: Annotated[
-        str | None, Query(pattern="^(coin_cash|invite_cash)$")
-    ] = None,
-) -> WithdrawSummaryOut:
-    return WithdrawSummaryOut(**queries.withdraw_summary(db, source=source))
+def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
+    return WithdrawSummaryOut(**queries.withdraw_summary(db))
 
 
 @router.get(
@@ -165,13 +153,13 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
     )
 
 
+@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
+def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
+    return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
+
+
 @router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
-def withdraw_detail(
-    out_bill_no: str,
-    db: AdminDb,
-    date_from: Annotated[datetime | None, Query()] = None,
-    date_to: Annotated[datetime | None, Query()] = None,
-) -> WithdrawDetailOut:
+def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
     order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
     if order is None:
         raise HTTPException(status_code=404, detail="提现单不存在")
@@ -184,8 +172,6 @@ def withdraw_detail(
             id=user.id,
             phone=user.phone,
             nickname=user.nickname,
-            is_high_risk=user.is_high_risk,
-            high_risk_note=user.high_risk_note,
             status=user.status,
             wechat_nickname=user.wechat_nickname,
             wechat_avatar_url=user.wechat_avatar_url,
@@ -209,36 +195,37 @@ def withdraw_detail(
         recent_withdraws,
         overview["cash_balance_cents"] if overview else 0,
     )
-    detail_enrichment = queries.withdraw_list_enrichment(
-        db, [order.user_id], source=order.source
-    ).get(order.user_id, {})
 
     return WithdrawDetailOut(
         order=WithdrawOrderOut.model_validate(order),
         user=user_snapshot,
-        cumulative_success_cents=int(
-            detail_enrichment.get("cumulative_success_cents", 0)
-        ),
         risk_flags=risk_flags,
         risk_score=risk_score,
         recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
         recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
         audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
-        invite_overview=(
-            InviteOverviewOut(
-                **queries.invite_overview(
-                    db,
-                    order.user_id,
-                    date_from=date_from,
-                    date_to=date_to,
-                )
-            )
-            if order.source == "invite_cash"
-            else None
-        ),
     )
 
 
+# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
+@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
+def reconcile(
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+    older_than_minutes: Annotated[int, Query(ge=0)] = 15,
+) -> ReconcileResult:
+    try:
+        result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
+    except wxpay.WxPayNotConfiguredError as e:
+        raise HTTPException(status_code=503, detail="微信支付未配置") from e
+    write_audit(
+        db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None,
+        detail=result, ip=get_client_ip(request), commit=True,
+    )
+    return ReconcileResult(**result)
+
+
 def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
     success = sum(1 for item in items if item.ok)
     return WithdrawBulkResult(

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,11 +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 | None = Field(None, description="按真实展示次数加权的 SDK eCPM(元/千次)")
+    ecpm_yuan: float = Field(..., description="按展示次数加权的 SDK eCPM(元/千次)")
 
 
 class AdRevenueRow(BaseModel):
@@ -101,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…);纯发奖行为空")
-    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
+    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);历史或未上报展示来源为空")
+    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/…;纯展示为空")
@@ -143,7 +143,7 @@ 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,
@@ -163,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/comparison.py

@@ -39,10 +39,8 @@ class AdminComparisonListItem(BaseModel):
     # 本次比价 LLM 总成本(元,按当时价冻结);旧记录/未回填为 None → 前端「成本」列回退估算。见 services/llm_cost.py。
     llm_cost_yuan: float | None = None
     device_model: str | None = None
-    device_model_name: str | None = None
     rom_vendor: str | None = None
     rom_name: str | None = None
-    rom_version: int | None = None
     android_version: str | None = None
     app_version: str | None = None
     ad_revenue_yuan: float = 0.0   # 本次比价看的信息流广告预估收益(元),queries 瞬态挂 ORM 实例上
@@ -86,6 +84,7 @@ class AdminComparisonDetail(AdminComparisonListItem):
     skipped_dish_names: list = []
     # 全量环境
     device_manufacturer: str | None = None
+    rom_version: int | None = None
     android_sdk: int | None = None
     app_version_code: int | None = None
     source_app_version: str | None = None

app/admin/schemas/feedback.py

@@ -32,10 +32,7 @@ class FeedbackOut(BaseModel):
     # 提交端环境快照(feedback 表列):提交版本号 / 机型OS版本;改版前的历史反馈为 None
     app_version: str | None = None
     device_model: str | None = None
-    device_model_name: str | None = None
-    device_manufacturer: str | None = None
     rom_name: str | None = None
-    rom_version: int | None = None
     android_version: str | None = None
     # 联表瞬态字段(queries._attach_user_info 挂):列表展示完整手机号,点手机号查该用户全部反馈
     phone: str | None = None

app/admin/schemas/user.py

@@ -16,8 +16,6 @@ class AdminUserListItem(BaseModel):
     register_channel: str
     status: str
     debug_trace_enabled: bool = False
-    is_high_risk: bool = False
-    high_risk_note: str | None = None
     wechat_openid: str | None = None
     wechat_nickname: str | None = None
     created_at: datetime
@@ -120,19 +118,3 @@ class SetUserStatusRequest(BaseModel):
 
 class SetDebugTraceRequest(BaseModel):
     enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")
-
-
-class SetUserRiskRequest(BaseModel):
-    is_high_risk: bool = Field(..., description="是否标记为高风险用户")
-    note: str | None = Field(
-        None,
-        max_length=500,
-        description="高风险备注；标记高风险时必填，解除后清空",
-    )
-
-    @field_validator("note")
-    @classmethod
-    def validate_note(cls, value: str | None, info):
-        if info.data.get("is_high_risk") and not (value or "").strip():
-            raise ValueError("标记高风险时必须填写原因")
-        return value.strip() if value and value.strip() else None

app/admin/schemas/wallet.py

@@ -60,8 +60,6 @@ class WithdrawListItemOut(WithdrawOrderOut):
 
     phone: str | None = None
     nickname: str | None = None
-    is_high_risk: bool = False
-    high_risk_note: str | None = None
     cumulative_success_cents: int = 0  # 累计成功提现 = SUM(amount_cents) WHERE status='success'
 
 
@@ -69,10 +67,7 @@ class WithdrawSummaryOut(BaseModel):
     reviewing_count: int
     reviewing_amount_cents: int
     pending_count: int
-    success_count: int
-    rejected_count: int
     failed_count: int
-    total_count: int
     today_success_count: int
     today_success_amount_cents: int
     today_rejected_count: int
@@ -82,8 +77,6 @@ class WithdrawUserSnapshot(BaseModel):
     id: int
     phone: str
     nickname: str | None = None
-    is_high_risk: bool = False
-    high_risk_note: str | None = None
     status: str
     wechat_nickname: str | None = None
     wechat_avatar_url: str | None = None
@@ -94,34 +87,19 @@ class WithdrawUserSnapshot(BaseModel):
     withdraw_success_cents: int
 
 
-class InviteeDetailOut(BaseModel):
-    user_id: int
-    phone: str
-    registered_at: datetime
-    invite_success: bool
-    first_compare_store: str | None = None
-    first_compare_products: str | None = None
-    first_order_store: str | None = None
-    first_order_products: str | None = None
-    first_order_amount_cents: int | None = None
-
-
-class InviteOverviewOut(BaseModel):
-    invite_total: int
-    invite_success_total: int
-    items: list[InviteeDetailOut]
-
-
 class WithdrawDetailOut(BaseModel):
     order: WithdrawOrderOut
     user: WithdrawUserSnapshot | None = None
-    cumulative_success_cents: int = 0
     risk_flags: list[str]
     risk_score: int
     recent_withdraws: list[WithdrawOrderOut]
     recent_cash_transactions: list[CashTxnOut]
     audit_logs: list[AdminAuditLogOut]
-    invite_overview: InviteOverviewOut | None = None
+
+
+class ReconcileResult(BaseModel):
+    checked: int
+    resolved: int
 
 
 class WithdrawBulkRequest(BaseModel):
@@ -150,6 +128,26 @@ class WithdrawBulkResult(BaseModel):
     items: list[WithdrawBulkItemResult]
 
 
+class WithdrawLedgerCheckOut(BaseModel):
+    ok: bool
+    # 普通现金账(coin_cash:金币兑换的现金)
+    cash_balance_total_cents: int
+    cash_transaction_total_cents: int
+    balance_diff_cents: int
+    missing_withdraw_txn_count: int
+    missing_refund_txn_count: int
+    duplicate_refund_txn_count: int
+    refund_txn_on_non_terminal_count: int
+    # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)。默认 0 向后兼容。
+    invite_cash_balance_total_cents: int = 0
+    invite_cash_transaction_total_cents: int = 0
+    invite_balance_diff_cents: int = 0
+    invite_missing_withdraw_txn_count: int = 0
+    invite_missing_refund_txn_count: int = 0
+    invite_duplicate_refund_txn_count: int = 0
+    invite_refund_txn_on_non_terminal_count: int = 0
+
+
 class WxpayHealthCheckOut(BaseModel):
     ok: bool
     wxpay_configured: bool

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/api/v1/compare.py

@@ -25,7 +25,7 @@ import uuid
 from typing import Any
 
 import httpx
-from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
+from fastapi import APIRouter, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import DbSession, OptionalUser
@@ -36,7 +36,6 @@ from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
-from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare")
 
@@ -81,7 +80,7 @@ def _harvest_running_blocking(
 def _harvest_done_blocking(
     trace_id: str, user_id: int | None, done_params: dict, business_type: str,
     device_id: str | None, device_info: dict | None, trace_url: str | None,
-) -> int:
+) -> None:
     with SessionLocal() as db:
         rec, newly_success = crud_compare.harvest_done(
             db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -103,7 +102,6 @@ def _harvest_done_blocking(
         # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
         # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
         # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
-        return rec.id
 
 
 def _harvest_abort_blocking(
@@ -245,12 +243,7 @@ async def intent_precoupon_step(
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(
-    request: Request,
-    background_tasks: BackgroundTasks,
-    user: OptionalUser,
-    db: DbSession,
-) -> dict[str, Any]:
+async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
@@ -259,15 +252,12 @@ async def price_step(
     if action.get("command") == "done" and not resp.get("continue", True):
         done_params = action.get("params") or {}
         try:
-            record_id = await run_in_threadpool(
+            await run_in_threadpool(
                 _harvest_done_blocking, trace_id, (user.id if user else None),
                 done_params, "food",
                 meta.get("device_id"), meta.get("device_info"),
                 resp.get("trace_url") or done_params.get("trace_url"),
             )
-            background_tasks.add_task(
-                backfill_comparison_llm_cost, record_id, trace_id
-            )
         except Exception as e:  # noqa: BLE001
             logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/compare_record.py

@@ -16,6 +16,8 @@ import logging
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
@@ -28,7 +30,8 @@ from app.schemas.compare_record import (
     ComparisonRecordOut,
     ComparisonRecordPage,
 )
-from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
+from app.services.llm_cost import compute_llm_cost, get_llm_prices
+from app.services.pricebot_llm_calls import fetch_llm_calls
 
 logger = logging.getLogger("shagua.compare_record")
 
@@ -118,7 +121,32 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
     """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
     独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    backfill_comparison_llm_cost(record_id, trace_id)
+    calls = fetch_llm_calls(trace_id)
+    if not calls:
+        return
+    db = SessionLocal()
+    try:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is None:
+            return
+        rec.llm_calls = calls
+        rec.llm_call_count = len(calls)
+        rec.retry_count = sum(1 for c in calls if c.get("error"))
+        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
+        # error 的调用 usage 可能为 None,or {} 兜底)
+        rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
+        rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
+        # 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
+        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
+        db.commit()
+        logger.info(
+            "backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
+            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
+        )
+    except Exception as e:  # noqa: BLE001 best-effort
+        logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
+    finally:
+        db.close()
 
 
 @router.get(

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)需主账号在「角色管理」授予「查看全部数据」权限,否则查不到
@@ -377,12 +377,6 @@ class Settings(BaseSettings):
     # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
     # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
     INTERNAL_API_SECRET: str = ""
-    # Current compare clients are persisted by server-side harvest. Repair any
-    # recent terminal rows left without LLM usage by transient upstream/auth failures.
-    LLM_COST_BACKFILL_ENABLED: bool = True
-    LLM_COST_BACKFILL_INTERVAL_SEC: int = 300
-    LLM_COST_BACKFILL_BATCH_SIZE: int = 100
-    LLM_COST_BACKFILL_LOOKBACK_DAYS: int = 30
 
     # ===== 媒体文件(用户头像上传)=====
     # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。

app/core/llm_cost_backfill_worker.py

@@ -1,63 +0,0 @@
-"""Periodic repair worker for comparison records with missing LLM token cost."""
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import logging
-
-from app.core.config import settings
-from app.services.comparison_llm_backfill import repair_missing_comparison_llm_costs
-from app.services.pricebot_llm_calls import pricebot_llm_auth_ready
-
-logger = logging.getLogger("shagua.llm_cost_backfill_worker")
-
-
-async def _run_loop() -> None:
-    interval = max(60, int(settings.LLM_COST_BACKFILL_INTERVAL_SEC))
-    logger.info(
-        "LLM cost backfill worker started interval=%ss batch=%s lookback_days=%s",
-        interval,
-        settings.LLM_COST_BACKFILL_BATCH_SIZE,
-        settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
-    )
-    try:
-        while True:
-            try:
-                auth_ready = await asyncio.to_thread(pricebot_llm_auth_ready)
-                if auth_ready:
-                    result = await asyncio.to_thread(
-                        repair_missing_comparison_llm_costs,
-                        limit=settings.LLM_COST_BACKFILL_BATCH_SIZE,
-                        lookback_days=settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
-                    )
-                    logger.info("LLM cost backfill batch result=%s", result)
-                else:
-                    logger.error(
-                        "LLM cost backfill skipped: PriceBot internal auth is not ready"
-                    )
-            except Exception:  # noqa: BLE001
-                logger.exception("LLM cost backfill batch failed")
-            await asyncio.sleep(interval)
-    except asyncio.CancelledError:
-        logger.info("LLM cost backfill worker stopped")
-        raise
-
-
-def start_llm_cost_backfill_worker() -> asyncio.Task | None:
-    if not settings.LLM_COST_BACKFILL_ENABLED:
-        logger.info("LLM cost backfill worker disabled")
-        return None
-    if not settings.INTERNAL_API_SECRET:
-        logger.warning(
-            "LLM cost backfill worker not started: INTERNAL_API_SECRET is empty"
-        )
-        return None
-    return asyncio.create_task(_run_loop(), name="llm-cost-backfill")
-
-
-async def stop_llm_cost_backfill_worker(task: asyncio.Task | None) -> None:
-    if task is None:
-        return
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task

app/integrations/pangle_report.py

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

app/main.py

@@ -60,10 +60,6 @@ from app.core.inactivity_reset_worker import (
     start_inactivity_reset_worker,
     stop_inactivity_reset_worker,
 )
-from app.core.llm_cost_backfill_worker import (
-    start_llm_cost_backfill_worker,
-    stop_llm_cost_backfill_worker,
-)
 from app.core.logging import setup_logging
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
@@ -107,7 +103,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
     inactivity_task = start_inactivity_reset_worker()
-    llm_cost_backfill_task = start_llm_cost_backfill_worker()
     try:
         yield
     finally:
@@ -117,7 +112,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
-        await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
         await aclose_pricebot_client()
         mt_meituan.close_client()
         logger.info("shutting down")

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`(客户端自报
-eCPM 折算的预估)互为对照:
++ `scripts/sync_pangle_revenue` 落库)，与 `ad_ecpm_record`(客户端 SDK eCPM 折算的预估)
+互为对照:
 
-- `revenue_yuan` ← 接口 `revenue`(预估收益,元;排序价×展示/1000,所有 ADN 都有);
-- `api_revenue_yuan` ← 接口 `api_revenue`(收益Api,元;各 ADN 经 Reporting 回传、更接近结算;
+- `revenue_yuan` ← 接口 `revenue`(排序价/竞价实时价预估,元,不是结算收入);
+- `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/user.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-from sqlalchemy import Boolean, DateTime, Integer, String, Text, false, func
+from sqlalchemy import Boolean, DateTime, Integer, String, false, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -62,13 +62,6 @@ class User(Base):
         Boolean, nullable=False, default=False, server_default=false()
     )
 
-    # 运营人工风险标记。与自动风控分值分开：这里表达人工复核结论，并保留可编辑备注，
-    # 供邀请提现、其他提现和用户管理三个页面统一展示。
-    is_high_risk: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=False, server_default=false(), index=True
-    )
-    high_risk_note: Mapped[str | None] = mapped_column(Text, nullable=True)
-
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False
     )

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:
-        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 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)]
+    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
     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(
-    results: list[dict], platform_results: dict | None = None
-) -> dict:
+def _derive_from_results(results: list[dict]) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _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
-
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
     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/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/repositories/wallet.py

@@ -6,7 +6,6 @@
 """
 from __future__ import annotations
 
-import logging
 import re
 import unicodedata
 import uuid
@@ -21,6 +20,7 @@ from app.core.config import settings
 from app.core.rewards import COIN_PER_CENT, coins_to_cents
 from app.integrations import wxpay
 from app.models.user import User
+from app.repositories.user import apply_wechat_display_identity
 from app.models.wallet import (
     CashTransaction,
     CoinAccount,
@@ -29,55 +29,12 @@ from app.models.wallet import (
     WechatTransferAuthorization,
     WithdrawOrder,
 )
-from app.repositories.user import apply_wechat_display_identity
 from app.services import notification_events
 
-logger = logging.getLogger(__name__)
-
 # 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
 _WX_STATE_SUCCESS = "SUCCESS"
 _WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
 _WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM"  # 用户还没在微信确认页确认
-_WX_FAIL_REASON_LABELS = {
-    "ACCOUNT_FROZEN": "用户微信账户被冻结",
-    "ACCOUNT_NOT_EXIST": "用户微信账户不存在",
-    "BANK_CARD_ACCOUNT_ABNORMAL": "用户银行卡已销户、冻结、作废或挂失",
-    "BANK_CARD_BANK_INFO_WRONG": "用户登记的银行或分支行信息有误",
-    "BANK_CARD_CARD_INFO_WRONG": "用户银行卡户名或卡号有误",
-    "BANK_CARD_COLLECTIONS_ABOVE_QUOTA": "用户银行卡收款达到限额",
-    "BANK_CARD_PARAM_ERROR": "用户收款银行卡信息错误",
-    "BANK_CARD_STATUS_ABNORMAL": "用户银行卡状态异常",
-    "BLOCK_B2C_USERLIMITAMOUNT_BSRULE_MONTH": "用户本月转账收款已达限额",
-    "BLOCK_B2C_USERLIMITAMOUNT_MONTH": "用户账户存在风险，本月收款受限",
-    "DAY_RECEIVED_COUNT_EXCEED": "用户当日收款次数已达上限",
-    "DAY_RECEIVED_QUOTA_EXCEED": "用户当日收款额度已达上限",
-    "EXCEEDED_ESTIMATED_AMOUNT": "转账金额超过预约金额范围",
-    "ID_CARD_NOT_CORRECT": "收款人身份证校验不通过",
-    "MCH_CANCEL": "商户已撤销付款",
-    "MERCHANT_REJECT": "商户转账验密人已驳回",
-    "MERCHANT_NOT_CONFIRM": "商户转账验密人超时未确认",
-    "NAME_NOT_CORRECT": "收款人姓名校验不通过",
-    "OPENID_INVALID": "用户 OpenID 无效或不属于当前 AppID",
-    "OTHER_FAIL_REASON_TYPE": "微信返回其他失败原因",
-    "OVERDUE_CLOSE": "超过微信系统重试期，订单自动关闭",
-    "PAYEE_ACCOUNT_ABNORMAL": "用户微信账户收款异常",
-    "PAYER_ACCOUNT_ABNORMAL": "商户账户付款受限",
-    "PRODUCT_AUTH_CHECK_FAIL": "商户未开通转账权限或权限已冻结",
-    "REALNAME_ACCOUNT_RECEIVED_QUOTA_EXCEED": "用户微信实名账户收款受限",
-    "REAL_NAME_CHECK_FAIL": "用户未完成微信实名认证",
-    "RECEIVE_ACCOUNT_NOT_CONFIGURE": "商户未配置收款用户列表",
-    "RESERVATION_INFO_NOT_MATCH": "转账信息与预约信息不一致",
-    "RESERVATION_SCENE_NOT_MATCH": "转账场景与预约场景不一致",
-    "RESERVATION_STATE_INVALID": "预约转账单状态异常",
-    "TRANSFER_QUOTA_EXCEED": "用户单笔收款额度已达上限",
-    "TRANSFER_REMARK_SET_FAIL": "微信转账备注设置失败",
-    "TRANSFER_RISK": "该笔转账存在风险，已被微信拦截",
-    "TRANSFER_SCENE_INVALID": "商户未获取当前转账场景",
-    "TRANSFER_SCENE_UNAVAILABLE": "当前转账场景暂不可用",
-    "RELATED_ORDER_TRANSFER_AMOUNT_EXCEED": "关联订单累计付款金额超过上限",
-    "RELATED_ORDER_TRANSFER_COUNT_EXCEED": "关联订单累计付款次数超过上限",
-    "BUDGET_NOT_ENOUGH": "商户预算资金不足",
-}
 # 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
 # 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提
 # (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。
@@ -522,19 +479,6 @@ def _try_deduct_cash(db: Session, user_id: int, amount_cents: int, source: str =
 
 def _add_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> int:
     """原子增加指定账户余额(退款用),返回加后余额。source 决定退回哪个账户(两账户隔离)。"""
-    # Normal withdrawals always create the account before deducting funds. This
-    # fallback covers legacy rows, hand-written fixtures, and broken migrations:
-    # a missing balance snapshot must not make a legitimate refund fail with 500.
-    if db.get(CoinAccount, user_id) is None:
-        logger.error(
-            "withdraw refund found missing coin_account; recreating empty account: "
-            "user_id=%s source=%s amount_cents=%s",
-            user_id,
-            source,
-            amount_cents,
-        )
-        get_or_create_account(db, user_id, commit=False)
-
     col = _balance_col(source)
     db.execute(
         update(CoinAccount)
@@ -638,32 +582,6 @@ def _wx_not_found(result: dict) -> bool:
     return "NOT_FOUND" in str(code)
 
 
-def _wechat_api_error_reason(data: object) -> str:
-    """Format a non-200 WeChat API response for operator display."""
-    if not isinstance(data, dict):
-        return f"微信发起转账失败：{data}"
-    code = str(data.get("code") or "").strip()
-    message = str(data.get("message") or "").strip()
-    if code and message:
-        return f"微信发起转账失败：{message}（{code}）"
-    return f"微信发起转账失败：{message or code or '未知错误'}"
-
-
-def _wechat_terminal_failure_reason(data: dict, state: str) -> str:
-    """Translate WeChat query ``fail_reason`` while preserving unknown codes."""
-    code = str(data.get("fail_reason") or "").strip()
-    if code:
-        label = _WX_FAIL_REASON_LABELS.get(code)
-        if label:
-            return f"微信转账失败：{label}（{code}）"
-        return f"微信转账失败：{code}"
-    if state == "CANCELLED":
-        return "微信转账已撤销（CANCELLED）"
-    if state == "CLOSED":
-        return "微信转账已关闭（CLOSED）"
-    return f"微信转账失败（状态：{state or 'FAIL'}）"
-
-
 def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> None:
     """转账调用结果不明(超时/异常/非200)时,**先查单再决定**,绝不盲目退款(防退款后又到账)。
       - 微信查到 SUCCESS        → 钱已出,置 success,不退款
@@ -689,17 +607,13 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
 
     state = q["data"].get("state", "")
     order.wechat_state = state
-    order.transfer_bill_no = q["data"].get("transfer_bill_no") or order.transfer_bill_no
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
+        order.transfer_bill_no = q["data"].get("transfer_bill_no")
         db.commit()
         notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
-        _refund_withdraw(
-            db,
-            order,
-            reason=_wechat_terminal_failure_reason(q["data"], state),
-        )
+        _refund_withdraw(db, order, reason=reason)
     else:
         order.package_info = q["data"].get("package_info") or order.package_info
         db.commit()
@@ -1053,14 +967,6 @@ def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> Wit
     order.package_info = data.get("package_info")  # 免确认转账无此字段(None);确认模式带它供拉确认页
     if data.get("state") == _WX_STATE_SUCCESS:
         order.status = "success"
-    elif data.get("state") in _WX_STATE_FAILED:
-        _refund_withdraw(
-            db,
-            order,
-            reason=_wechat_terminal_failure_reason(data, str(data.get("state") or "")),
-        )
-        db.refresh(order)
-        return order
     db.commit()
     db.refresh(order)
     if order.status == "success":  # 免确认转账直接到账 → PRD #3 提现到账
@@ -1105,11 +1011,7 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
             return order
         if result["status_code"] != 200:
             # 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态)
-            _settle_after_ambiguous(
-                db,
-                order,
-                reason=_wechat_api_error_reason(result["data"]),
-            )
+            _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
             # 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权
             _refresh_active_auth(db, order.user_id)
             db.refresh(order)
@@ -1137,11 +1039,7 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
         return order
 
     if result["status_code"] != 200:
-        _settle_after_ambiguous(
-            db,
-            order,
-            reason=_wechat_api_error_reason(result["data"]),
-        )
+        _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
         db.refresh(order)
         return order
 
@@ -1198,24 +1096,19 @@ def refresh_withdraw_status(
     ).scalar_one_or_none()
     if order is None:
         raise WithdrawOrderNotFound
-    if order.status not in {"pending", "failed"}:
-        return order
-    enrich_failed_order = order.status == "failed"
+    if order.status != "pending":
+        return order  # 已终态,不再查
 
     try:
         result = wxpay.query_transfer(out_bill_no)
     except wxpay.WxPayNotConfiguredError:
         raise
     except Exception as exc:  # noqa: BLE001 - 查单失败不能把运营后台打成 500
-        if enrich_failed_order:
-            return order
         order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256]
         db.commit()
         db.refresh(order)
         return order
     if result["status_code"] != 200:
-        if enrich_failed_order:
-            return order
         if _wx_not_found(result):
             # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全
             _refund_withdraw(db, order, reason="微信无此单,已退回")
@@ -1224,32 +1117,12 @@ def refresh_withdraw_status(
 
     state = result["data"].get("state", "")
     order.wechat_state = state
-    order.transfer_bill_no = (
-        result["data"].get("transfer_bill_no") or order.transfer_bill_no
-    )
-    if enrich_failed_order:
-        if state in _WX_STATE_FAILED:
-            order.fail_reason = _wechat_terminal_failure_reason(
-                result["data"], state
-            )[:256]
-        elif state == _WX_STATE_SUCCESS:
-            order.fail_reason = (
-                "资金状态异常：本地已退款，但微信查单显示已到账，请人工核查"
-            )
-        db.commit()
-        db.refresh(order)
-        return order
-
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
         db.commit()
         notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
-        _refund_withdraw(
-            db,
-            order,
-            reason=_wechat_terminal_failure_reason(result["data"], state),
-        )
+        _refund_withdraw(db, order, reason=f"微信转账状态 {state}")
     elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed:
         # 用户从确认页回来了却仍未确认 → 视为放弃:撤销微信单(防事后确认导致重复打款)后退款。
         # 撤单失败(可能已被确认进 ACCEPTED 的竞态)则保持 pending,等下次查询。

app/schemas/compare_record.py

@@ -85,10 +85,6 @@ class ComparisonResultIn(BaseModel):
     status: str | None = None
     # 门店打烊原因(price 为 None 时带): 与 status="store_closed" 等价的更早信号, 一并落库供前端兜底判打烊。
     store_closed: str | None = None
-    # 该平台缺菜(漏菜)数量(pricebot 冗余进 comparison_results 行): 有价但少买了菜时 >0。
-    # 记录页三平台网格据此逐格标"缺少 X 个菜品"(网格只拿 comparison_results, 拿不到 platform_results)。
-    # 必须显式声明: 落库走 model_dump(), 不声明会被 pydantic 静默丢弃 → 记录页拿不到数量。
-    skipped_dish_count: int | None = None
 
 
 class ComparisonRecordIn(BaseModel):

app/services/comparison_llm_backfill.py

@@ -1,156 +0,0 @@
-"""Persist and repair comparison-record LLM token costs."""
-from __future__ import annotations
-
-import logging
-import time
-from datetime import UTC, datetime, timedelta
-
-from sqlalchemy import select
-
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal
-from app.models.app_config import AppConfig
-from app.models.comparison import ComparisonRecord
-from app.services.llm_cost import compute_llm_cost, get_llm_prices
-from app.services.pricebot_llm_calls import fetch_llm_calls
-
-logger = logging.getLogger("shagua.comparison_llm_backfill")
-
-
-def _utc_to_beijing_naive(value: datetime) -> datetime:
-    """Convert a DB UTC timestamp to comparison_record's Beijing wall-clock."""
-    if value.tzinfo is None:
-        value = value.replace(tzinfo=UTC)
-    return value.astimezone(CN_TZ).replace(tzinfo=None)
-
-
-def _store_calls(record_id: int, trace_id: str, calls: list[dict]) -> bool:
-    """Store calls and all derived fields atomically."""
-    with SessionLocal() as db:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is None or rec.trace_id != trace_id:
-            logger.warning(
-                "LLM cost backfill record mismatch record_id=%s trace=%s",
-                record_id,
-                trace_id,
-            )
-            return False
-
-        # Never recalculate a frozen historical cost with a newer price config.
-        if rec.llm_cost_yuan is not None and rec.llm_calls:
-            return False
-
-        rec.llm_calls = calls
-        rec.llm_call_count = len(calls)
-        rec.retry_count = sum(1 for call in calls if call.get("error"))
-        rec.input_tokens = sum(
-            (call.get("usage") or {}).get("prompt_tokens") or 0 for call in calls
-        )
-        rec.output_tokens = sum(
-            (call.get("usage") or {}).get("completion_tokens") or 0 for call in calls
-        )
-        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(
-            calls, get_llm_prices(db)
-        )
-        db.commit()
-        logger.info(
-            "LLM cost backfilled trace=%s calls=%d input_tokens=%d "
-            "output_tokens=%d cost=%s",
-            trace_id,
-            len(calls),
-            rec.input_tokens,
-            rec.output_tokens,
-            rec.llm_cost_yuan,
-        )
-        return True
-
-
-def backfill_comparison_llm_cost(
-    record_id: int,
-    trace_id: str,
-    *,
-    attempts: int = 3,
-    retry_delays: tuple[float, ...] = (1.0, 3.0),
-) -> bool:
-    """Fetch and persist one record, retrying short-lived upstream races."""
-    if not settings.INTERNAL_API_SECRET or not trace_id:
-        logger.warning(
-            "LLM cost backfill skipped trace=%s: INTERNAL_API_SECRET is not configured",
-            trace_id,
-        )
-        return False
-    total_attempts = max(1, attempts)
-    for attempt in range(total_attempts):
-        calls = fetch_llm_calls(trace_id)
-        if calls:
-            try:
-                return _store_calls(record_id, trace_id, calls)
-            except Exception:  # noqa: BLE001 - background repair must stay alive
-                logger.exception(
-                    "LLM cost store failed trace=%s record_id=%s",
-                    trace_id,
-                    record_id,
-                )
-                return False
-
-        if attempt + 1 < total_attempts:
-            delay = retry_delays[min(attempt, len(retry_delays) - 1)] if retry_delays else 0
-            if delay > 0:
-                time.sleep(delay)
-
-    logger.warning(
-        "LLM cost backfill has no calls trace=%s record_id=%s attempts=%d",
-        trace_id,
-        record_id,
-        total_attempts,
-    )
-    return False
-
-
-def repair_missing_comparison_llm_costs(
-    *,
-    limit: int = 100,
-    lookback_days: int = 30,
-) -> dict[str, int]:
-    """Repair a bounded batch of recent terminal records with missing cost."""
-    cutoff = datetime.now(CN_TZ).replace(tzinfo=None) - timedelta(
-        days=max(1, lookback_days)
-    )
-    with SessionLocal() as db:
-        # app_config has no price history. Repricing a record from before the
-        # current config became effective would fabricate a historical cost, so
-        # only repair records at/after that timestamp.
-        price_config_updated_at = db.execute(
-            select(AppConfig.updated_at).where(AppConfig.key == "llm_token_price")
-        ).scalar_one_or_none()
-        date_conditions = [ComparisonRecord.created_at >= cutoff]
-        if price_config_updated_at is not None:
-            date_conditions.append(
-                ComparisonRecord.created_at
-                >= _utc_to_beijing_naive(price_config_updated_at)
-            )
-        candidates = list(
-            db.execute(
-                select(ComparisonRecord.id, ComparisonRecord.trace_id)
-                .where(
-                    *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed")),
-                    ComparisonRecord.llm_cost_yuan.is_(None),
-                )
-                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
-                .limit(max(1, limit))
-            ).all()
-        )
-
-    repaired = 0
-    for record_id, trace_id in candidates:
-        if backfill_comparison_llm_cost(
-            record_id, trace_id, attempts=1, retry_delays=()
-        ):
-            repaired += 1
-    return {
-        "candidates": len(candidates),
-        "repaired": repaired,
-        "unresolved": len(candidates) - repaired,
-    }

app/services/pricebot_llm_calls.py

@@ -20,67 +20,19 @@ from app.core.pricebot_router import pick_pricebot
 logger = logging.getLogger("shagua.pricebot_llm")
 
 
-def pricebot_llm_auth_ready() -> bool:
-    """Verify every configured PriceBot instance accepts the shared secret."""
-    secret = settings.INTERNAL_API_SECRET
-    if not secret:
-        logger.error("PriceBot LLM auth check failed: INTERNAL_API_SECRET is empty")
-        return False
-    for base in settings.pricebot_instances:
-        url = f"{base.rstrip('/')}/api/internal/llm_calls/__auth_probe__"
-        try:
-            resp = httpx.get(
-                url, headers={"X-Internal-Secret": secret}, timeout=3.0
-            )
-        except Exception as exc:  # noqa: BLE001
-            logger.error("PriceBot LLM auth check unavailable base=%s: %s", base, exc)
-            return False
-        if resp.status_code != 200:
-            logger.error(
-                "PriceBot LLM auth check rejected base=%s status=%s; "
-                "verify both services use the same INTERNAL_API_SECRET",
-                base,
-                resp.status_code,
-            )
-            return False
-    return True
-
-
 def fetch_llm_calls(trace_id: str) -> list[dict]:
     """返回该次比价的 LLM 调用明细列表(每条 {scene,model,input_messages,output,usage,latency_ms,error});
     未配密钥 / 无 trace_id / 拉取失败 → []。"""
     secret = settings.INTERNAL_API_SECRET
     if not secret or not trace_id:
         return []
-    preferred = pick_pricebot(trace_id)
-    # LLM JSONL is instance-local. If the cluster topology changed after a
-    # historical trace was created, consistent hashing may now point elsewhere;
-    # probe the remaining configured instances only when the preferred one is empty.
-    bases = [preferred, *(base for base in settings.pricebot_instances if base != preferred)]
-    for base in bases:
-        url = f"{base.rstrip('/')}/api/internal/llm_calls/{trace_id}"
-        try:
-            resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
-            if resp.status_code == 200:
-                calls = resp.json().get("calls", []) or []
-                if calls:
-                    return calls
-                continue
-            if resp.status_code in (401, 403):
-                logger.error(
-                    "fetch_llm_calls rejected trace=%s base=%s status=%s; "
-                    "INTERNAL_API_SECRET differs between app-server and PriceBot",
-                    trace_id,
-                    base,
-                    resp.status_code,
-                )
-            else:
-                logger.warning(
-                    "fetch_llm_calls trace=%s base=%s status=%s",
-                    trace_id,
-                    base,
-                    resp.status_code,
-                )
-        except Exception as e:  # noqa: BLE001 — best-effort
-            logger.warning("fetch_llm_calls trace=%s base=%s failed: %s", trace_id, base, e)
+    base = pick_pricebot(trace_id).rstrip("/")
+    url = f"{base}/api/internal/llm_calls/{trace_id}"
+    try:
+        resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
+        if resp.status_code == 200:
+            return resp.json().get("calls", []) or []
+        logger.warning("fetch_llm_calls trace=%s status=%s", trace_id, resp.status_code)
+    except Exception as e:  # noqa: BLE001 — best-effort,任何异常都不该影响上报
+        logger.warning("fetch_llm_calls trace=%s failed: %s", trace_id, e)
     return []

deploy/comparison-llm-cost.md

@@ -1,40 +0,0 @@
-# 比价 TOKEN 成本采集与补偿
-
-## 部署前置
-
-App Server 与 PriceBot 使用各自独立的 `.env`，但下面的值必须完全一致：
-
-- `/opt/shaguabijia-app-server/.env`
-- `/opt/pricebot-backend/.env`
-- 配置项：`INTERNAL_API_SECRET`
-
-不要把密钥原文写入日志、命令历史或 Git。修改后同时重启两个服务。
-
-App Server 启动后会逐个探测 `PRICEBOT_INSTANCES` 的内部读取接口。鉴权不一致时会记录
-`PriceBot LLM auth check rejected`，并跳过本轮补偿，避免对所有缺失记录重复发送失败请求。
-
-## 数据链路
-
-1. 当前客户端由 App Server 在 PriceBot 最终 `done` 帧到达时 harvest 比价记录。
-2. harvest 成功后立即异步读取同一 `trace_id` 的 LLM 调用，冻结 Token、成本和单价快照。
-3. 周期 worker 扫描近期 `success/failed` 且 `llm_cost_yuan IS NULL` 的记录进行补偿；
-   为避免用现价伪造历史成本，只处理当前单价配置生效时间之后的记录。
-4. 管理后台顶部“平均 TOKEN 成本”使用筛选范围内已冻结成本的数据库平均值。
-
-## 上线验收（只读 SQL）
-
-```sql
-SELECT
-  (created_at AT TIME ZONE 'Asia/Shanghai')::date AS day,
-  count(*) AS records,
-  count(llm_cost_yuan) AS cost_records,
-  round(avg(llm_cost_yuan)::numeric, 6) AS avg_token_cost
-FROM comparison_record
-WHERE created_at >= now() - interval '3 days'
-GROUP BY 1
-ORDER BY 1 DESC;
-```
-
-新产生的正常终态比价记录应在短时间内写入 `input_tokens`、`output_tokens` 和
-`llm_cost_yuan`。历史记录只有在 PriceBot 的对应 trace JSONL 仍保留时才能准确回填；
-原始调用已经清理的记录不能用估算值冒充真实成本。

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 天。
-- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(预估)+ `api_revenue`(结算口径)。
+- 每天 10:30 拉初值、14:30 拉日终值，均由 `scripts/sync_pangle_revenue.py` 以 `--days 3` 回补近 3 天。
+- 维度 = 日期 × 应用(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 触发。
-# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的「穿山甲后台收益(T+1)」区块。
+# 拉 GroMore T+1 天级收益入库 —— 单轮跑,由 timer 每天 10:30、14:30 触发。
+# 落 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/seed_mock_feedback.py

@@ -175,22 +175,22 @@ def seed(db) -> list[Feedback]:
         # 普通反馈 · 新端(带环境快照)· 无图
         fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
            source="profile",
-           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS 14", android_version="14",
+           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
            created=ago(minutes=6)),
         # 比价反馈 · scene=优惠不对 · 新端 · 2 图
         fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
            source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
-           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
+           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
            created=ago(minutes=22)),
         # 比价反馈 · scene=找错商品 · 新端 · 无图
         fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
            source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", android_version="14",
+           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
            created=ago(hours=1)),
         # 普通反馈 · 新端 · 1 图(表扬 + 小问题)
         fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
            source="profile", images=[imgs[2]],
-           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI 14", android_version="14",
+           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
            created=ago(hours=3)),
         # 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
         fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
@@ -206,7 +206,7 @@ def seed(db) -> list[Feedback]:
            source="profile", images=[imgs[3]],
            status="adopted", reward_coins=2000,
            review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
-           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS 13", android_version="13",
+           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
            created=ago(days=2)),
 
         # ===== 未采纳 rejected(带原因 + 回复)=====
@@ -214,7 +214,7 @@ def seed(db) -> list[Feedback]:
            source="comparison", scene="价格不准",
            status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
            admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
-           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
+           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
            created=ago(days=3)),
     ]
     db.add_all(feedbacks)

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 跑一次
-(默认拉【昨天】);穿山甲对历史数据可能订正,故支持回补近 N 天(幂等 upsert,重跑无害)。
+GroMore 数据 API 为 T+1:次日约 10:00 出初值，第三方 ADN Reporting 最晚约 13:50 更新。
+线上由 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,
-            "too_short": 0.0,
-            "capped": 0.1,
+            "closed_early": 0.1,
+            "too_short": 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 sum(row["revenue_yuan"] for row in result["hourly"]) == 0.2
+        assert result["daily"][0]["revenue_yuan"] == 1.3
+        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,7 +234,7 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         db.close()
 
 
-def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
+def test_category_stats_merge_legacy_feed_and_withdrawal_video() -> None:
     db = SessionLocal()
     phone = "18800009993"
     category_date = "2040-02-05"
@@ -220,6 +243,7 @@ def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
         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",
@@ -230,6 +254,7 @@ def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
                 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",
@@ -264,54 +289,78 @@ def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
         db.close()
 
 
-def test_feed_reward_details_keep_each_record_adn() -> None:
+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,
-                reward_date=detail_date, duration_seconds=20, 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",
+                client_event_id="detail-adn-pangle",
+                user_id=user.id,
+                reward_date=DETAIL_DATE,
+                duration_seconds=20,
+                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,
-                reward_date=detail_date, duration_seconds=20, 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",
+                client_event_id="detail-adn-gdt",
+                user_id=user.id,
+                reward_date=DETAIL_DATE,
+                duration_seconds=20,
+                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,
-            user_id=user.id, app_env="prod", revenue_scope="all",
+            db,
+            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)
@@ -319,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",
@@ -327,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",
@@ -336,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,
-            user_id=user.id, app_env="prod", revenue_scope="all",
+            db,
+            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"]}
@@ -366,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(AdEcpmRecord).where(AdEcpmRecord.report_date == fallback_date))
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == SOURCE_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_feedback_device_details.py

@@ -1,71 +0,0 @@
-from datetime import datetime, timedelta, timezone
-
-from app.admin.repositories import queries
-from app.db.session import SessionLocal
-from app.models.comparison import ComparisonRecord
-from app.models.feedback import Feedback
-from app.repositories import user as user_repo
-
-
-def test_feedback_device_details_use_same_user_and_model() -> None:
-    with SessionLocal() as db:
-        submitted_at = datetime.now(timezone.utc)
-        user = user_repo.upsert_user_for_login(
-            db,
-            phone="13800009876",
-            register_channel="sms",
-        )
-        db.add_all(
-            [
-                ComparisonRecord(
-                    user_id=user.id,
-                    trace_id="feedback-device-v2166ba",
-                    device_model="V2166BA",
-                    device_manufacturer="vivo",
-                    rom_name="OriginOS",
-                    rom_version=13,
-                    created_at=submitted_at - timedelta(minutes=5),
-                ),
-                ComparisonRecord(
-                    user_id=user.id,
-                    trace_id="feedback-device-other",
-                    device_model="OTHER-CODE",
-                    device_manufacturer="Other",
-                    rom_name="OtherOS",
-                    rom_version=99,
-                    created_at=submitted_at - timedelta(minutes=5),
-                ),
-                ComparisonRecord(
-                    user_id=user.id,
-                    trace_id="feedback-device-future-upgrade",
-                    device_model="V2166BA",
-                    device_manufacturer="vivo-new",
-                    rom_name="OriginOS",
-                    rom_version=99,
-                    created_at=submitted_at + timedelta(minutes=5),
-                ),
-                Feedback(
-                    user_id=user.id,
-                    content="设备信息补全测试",
-                    contact="",
-                    status="pending",
-                    device_model="V2166BA",
-                    rom_name="OriginOS",
-                    android_version="13",
-                    created_at=submitted_at,
-                ),
-            ]
-        )
-        db.commit()
-
-        items, _next_cursor, total = queries.list_feedbacks(
-            db,
-            user_id=user.id,
-            limit=20,
-        )
-
-        assert total == 1
-        assert items[0].device_model == "V2166BA"
-        assert items[0].device_model_name == "vivo Y77e"
-        assert items[0].device_manufacturer == "vivo"
-        assert items[0].rom_version == 13

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,13 +9,9 @@ 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
-from app.models.invite import InviteRelation
-from app.models.savings import SavingsRecord
-from app.models.user import User
 from app.models.wallet import CashTransaction, WithdrawOrder
 from app.repositories import user as user_repo
 from app.repositories import wallet as wallet_repo
@@ -148,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:
@@ -168,68 +162,6 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     assert records.status_code == 200, records.text
 
 
-def test_user_reward_stats_can_scope_withdrawals_by_account(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    uid = _seed_user_with_data("13800000023")
-    db = SessionLocal()
-    try:
-        account = wallet_repo.get_or_create_account(db, uid)
-        account.cash_balance_cents = 123
-        account.invite_cash_balance_cents = 456
-        db.add_all(
-            [
-                WithdrawOrder(
-                    user_id=uid,
-                    out_bill_no="rewardstatsinvite00000001",
-                    amount_cents=250,
-                    source="invite_cash",
-                    status="success",
-                ),
-                WithdrawOrder(
-                    user_id=uid,
-                    out_bill_no="rewardstatsinvite00000002",
-                    amount_cents=50,
-                    source="invite_cash",
-                    status="reviewing",
-                ),
-            ]
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    coin = admin_client.get(
-        f"/admin/api/users/{uid}/reward-stats",
-        params={"withdraw_source": "coin_cash"},
-        headers=_auth(admin_token),
-    )
-    invite = admin_client.get(
-        f"/admin/api/users/{uid}/reward-stats",
-        params={"withdraw_source": "invite_cash"},
-        headers=_auth(admin_token),
-    )
-    assert coin.status_code == 200, coin.text
-    assert invite.status_code == 200, invite.text
-    assert coin.json()["withdraw_success_cents"] == 100
-    assert coin.json()["withdraw_total"] == 1
-    assert coin.json()["cash_balance_cents"] == 123
-    assert invite.json()["withdraw_success_cents"] == 250
-    assert invite.json()["withdraw_total"] == 2
-    assert invite.json()["cash_balance_cents"] == 456
-
-
-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))
@@ -375,19 +307,10 @@ def test_withdraw_list_exposes_source_and_filters(
     db = SessionLocal()
     try:
         uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id
-        # 直接从 ORM 查询，验证列表富化返回人工风险结论。
-        user = db.get(User, uid)
-        assert user is not None
-        user.is_high_risk = True
-        user.high_risk_note = "邀请行为异常"
         db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001",
                              amount_cents=100, status="success", source="coin_cash"))
         db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001",
                              amount_cents=200, status="success", source="invite_cash"))
-        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview",
-                             amount_cents=50, status="reviewing", source="coin_cash"))
-        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview",
-                             amount_cents=80, status="reviewing", source="invite_cash"))
         db.commit()
     finally:
         db.close()
@@ -402,242 +325,6 @@ def test_withdraw_list_exposes_source_and_filters(
     )
     items = r.json()["items"]
     assert items and all(it["source"] == "invite_cash" for it in items)
-    assert all(it["is_high_risk"] is True for it in items)
-    assert all(it["high_risk_note"] == "邀请行为异常" for it in items)
-    # 累计提现按当前审核页账户隔离，不把普通现金的 100 分串进邀请金。
-    assert all(it["cumulative_success_cents"] == 200 for it in items)
-
-    invite_summary = admin_client.get(
-        "/admin/api/withdraws/summary",
-        params={"source": "invite_cash"},
-        headers=_auth(admin_token),
-    ).json()
-    other_summary = admin_client.get(
-        "/admin/api/withdraws/summary",
-        params={"source": "coin_cash"},
-        headers=_auth(admin_token),
-    ).json()
-    assert invite_summary["reviewing_count"] >= 1
-    assert other_summary["reviewing_count"] >= 1
-    assert invite_summary["reviewing_amount_cents"] >= 80
-    assert other_summary["reviewing_amount_cents"] >= 50
-
-
-def test_invite_withdraw_detail_contains_invitee_first_actions(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    db = SessionLocal()
-    try:
-        inviter = user_repo.upsert_user_for_login(
-            db, phone="13800009031", register_channel="sms"
-        )
-        invitee = user_repo.upsert_user_for_login(
-            db, phone="13800009032", register_channel="sms"
-        )
-        db.add(
-            InviteRelation(
-                inviter_user_id=inviter.id,
-                invitee_user_id=invitee.id,
-                status="effective",
-                compare_reward_granted=True,
-                compare_reward_cents=50,
-            )
-        )
-        db.add(
-            ComparisonRecord(
-                user_id=invitee.id,
-                trace_id="invite-detail-first-compare",
-                status="success",
-                store_name="首次比价商家",
-                product_names="商品甲、商品乙",
-            )
-        )
-        db.add(
-            SavingsRecord(
-                user_id=invitee.id,
-                order_amount_cents=1888,
-                saved_amount_cents=300,
-                source="compare",
-                shop_name="首次下单商家",
-                title="首次下单商品",
-                dishes=[],
-            )
-        )
-        bill = "inviteoverviewbill0001"
-        db.add(
-            WithdrawOrder(
-                user_id=inviter.id,
-                out_bill_no=bill,
-                amount_cents=100,
-                status="reviewing",
-                source="invite_cash",
-            )
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    response = admin_client.get(
-        f"/admin/api/withdraws/{bill}", headers=_auth(admin_token)
-    )
-    assert response.status_code == 200, response.text
-    overview = response.json()["invite_overview"]
-    assert overview["invite_total"] == 1
-    assert overview["invite_success_total"] == 1
-    item = overview["items"][0]
-    assert item["phone"] == "13800009032"
-    assert item["first_compare_store"] == "首次比价商家"
-    assert item["first_compare_products"] == "商品甲、商品乙"
-    assert item["first_order_store"] == "首次下单商家"
-    assert item["first_order_products"] == "首次下单商品"
-    assert item["first_order_amount_cents"] == 1888
-
-    filtered = admin_client.get(
-        f"/admin/api/withdraws/{bill}",
-        params={
-            "date_from": "2099-01-01T00:00:00Z",
-            "date_to": "2099-01-02T00:00:00Z",
-        },
-        headers=_auth(admin_token),
-    )
-    assert filtered.status_code == 200, filtered.text
-    assert filtered.json()["invite_overview"]["invite_total"] == 0
-
-
-def test_withdraw_summary_counts_match_status_lists_by_source(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """每个审核页签的汇总数必须等于相同 source、status 的列表总数。"""
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(
-            db, phone="13800009003", register_channel="sms"
-        ).id
-        for index, status in enumerate(
-            ("reviewing", "pending", "success", "rejected", "failed"), start=1
-        ):
-            db.add(
-                WithdrawOrder(
-                    user_id=uid,
-                    out_bill_no=f"summary{uid}invite{index}",
-                    amount_cents=index * 10,
-                    status=status,
-                    source="invite_cash",
-                )
-            )
-        db.add(
-            WithdrawOrder(
-                user_id=uid,
-                out_bill_no=f"summary{uid}coinfailed",
-                amount_cents=99,
-                status="failed",
-                source="coin_cash",
-            )
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    status_fields = {
-        "reviewing": "reviewing_count",
-        "pending": "pending_count",
-        "success": "success_count",
-        "rejected": "rejected_count",
-        "failed": "failed_count",
-    }
-    for source in ("coin_cash", "invite_cash"):
-        summary_response = admin_client.get(
-            "/admin/api/withdraws/summary",
-            params={"source": source},
-            headers=_auth(admin_token),
-        )
-        assert summary_response.status_code == 200, summary_response.text
-        summary = summary_response.json()
-
-        for status, field in status_fields.items():
-            list_response = admin_client.get(
-                "/admin/api/withdraws",
-                params={"source": source, "status": status, "limit": 1},
-                headers=_auth(admin_token),
-            )
-            assert list_response.status_code == 200, list_response.text
-            assert summary[field] == list_response.json()["total"]
-
-        reviewing_response = admin_client.get(
-            "/admin/api/withdraws",
-            params={"source": source, "status": "reviewing", "limit": 100},
-            headers=_auth(admin_token),
-        )
-        assert reviewing_response.status_code == 200, reviewing_response.text
-        reviewing_body = reviewing_response.json()
-        assert reviewing_body["total"] == len(reviewing_body["items"])
-        assert summary["reviewing_amount_cents"] == sum(
-            item["amount_cents"] for item in reviewing_body["items"]
-        )
-
-        all_response = admin_client.get(
-            "/admin/api/withdraws",
-            params={"source": source, "limit": 1},
-            headers=_auth(admin_token),
-        )
-        assert all_response.status_code == 200, all_response.text
-        assert summary["total_count"] == all_response.json()["total"]
-        assert summary["total_count"] == sum(
-            summary[field] for field in status_fields.values()
-        )
-
-    assert admin_client.get(
-        "/admin/api/withdraws/summary",
-        params={"source": "unknown"},
-        headers=_auth(admin_token),
-    ).status_code == 422
-
-
-def test_comparison_records_show_readable_device_and_rom_version(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    db = SessionLocal()
-    try:
-        user = user_repo.upsert_user_for_login(
-            db, phone="13800009041", register_channel="sms"
-        )
-        record = ComparisonRecord(
-            user_id=user.id,
-            trace_id="comparison-readable-device",
-            status="success",
-            device_model="V2166BA",
-            device_manufacturer="vivo",
-            rom_vendor="vivo",
-            rom_name="OriginOS",
-            rom_version=4,
-            android_version="14",
-        )
-        db.add(record)
-        db.commit()
-        record_id = record.id
-        user_id = user.id
-    finally:
-        db.close()
-
-    response = admin_client.get(
-        "/admin/api/comparison-records",
-        params={"user_id": user_id},
-        headers=_auth(admin_token),
-    )
-    assert response.status_code == 200, response.text
-    item = response.json()["items"][0]
-    assert item["device_model"] == "V2166BA"
-    assert item["device_model_name"] == "vivo Y77e"
-    assert item["rom_name"] == "OriginOS"
-    assert item["rom_version"] == 4
-
-    detail = admin_client.get(
-        f"/admin/api/comparison-records/{record_id}",
-        headers=_auth(admin_token),
-    )
-    assert detail.status_code == 200, detail.text
-    assert detail.json()["device_model_name"] == "vivo Y77e"
-    assert detail.json()["rom_version"] == 4
 
 
 def test_ad_coin_audit_full_count_truncate_and_only_mismatch(

tests/test_admin_roles.py

@@ -190,13 +190,7 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     assert roles["finance"]["label"] == "财务"
     assert roles["tech"]["label"] == "技术"
     # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
-    assert set(roles["finance"]["pages"]) == {
-        "dashboard",
-        "ad-revenue-report",
-        "cps",
-        "invite-withdraws",
-        "withdraws",
-    }
+    assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
     assert set(roles["tech"]["pages"]) == {
         "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
         "ad-revenue", "huawei-review", "event-logs", "audit-logs",

tests/test_admin_write.py

@@ -103,61 +103,6 @@ def _seed_price_report(phone: str) -> int:
         db.close()
 
 
-def test_set_user_risk_requires_note_and_writes_audit(
-    admin_client: TestClient, operator_token: str
-) -> None:
-    uid = _seed_user("13900000991")
-
-    missing_note = admin_client.post(
-        f"/admin/api/users/{uid}/risk",
-        json={"is_high_risk": True, "note": "   "},
-        headers=_auth(operator_token),
-    )
-    assert missing_note.status_code == 422
-
-    marked = admin_client.post(
-        f"/admin/api/users/{uid}/risk",
-        json={"is_high_risk": True, "note": "邀请记录存在批量异常"},
-        headers=_auth(operator_token),
-    )
-    assert marked.status_code == 200, marked.text
-
-    listed = admin_client.get(
-        "/admin/api/users",
-        params={"phone": "13900000991"},
-        headers=_auth(operator_token),
-    ).json()["items"][0]
-    assert listed["is_high_risk"] is True
-    assert listed["high_risk_note"] == "邀请记录存在批量异常"
-
-    db = SessionLocal()
-    try:
-        audit = db.execute(
-            select(AdminAuditLog)
-            .where(
-                AdminAuditLog.action == "user.risk.set",
-                AdminAuditLog.target_id == str(uid),
-            )
-            .order_by(AdminAuditLog.id.desc())
-        ).scalars().first()
-        assert audit is not None
-        assert audit.detail["after"]["is_high_risk"] is True
-    finally:
-        db.close()
-
-    cleared = admin_client.post(
-        f"/admin/api/users/{uid}/risk",
-        json={"is_high_risk": False, "note": None},
-        headers=_auth(operator_token),
-    )
-    assert cleared.status_code == 200
-    overview = admin_client.get(
-        f"/admin/api/users/{uid}", headers=_auth(operator_token)
-    ).json()
-    assert overview["user"]["is_high_risk"] is False
-    assert overview["user"]["high_risk_note"] is None
-
-
 # ===== 调金币 =====
 
 def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
@@ -634,47 +579,15 @@ def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> N
     ).status_code == 404
 
 
-def test_withdraw_refresh_enriches_existing_failed_reason(
-    admin_client: TestClient, finance_token: str, monkeypatch
-) -> None:
-    uid = _seed_user("13900000017")
-    with SessionLocal() as db:
-        db.add(
-            WithdrawOrder(
-                user_id=uid,
-                out_bill_no="adminfailedreason01",
-                amount_cents=100,
-                status="failed",
-                wechat_state="FAIL",
-                fail_reason="微信转账状态 FAIL",
-            )
-        )
-        db.commit()
-
+def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
     from app.repositories import wallet as wr
-
     monkeypatch.setattr(
-        wr.wxpay,
-        "query_transfer",
-        lambda out_bill_no: {
-            "status_code": 200,
-            "data": {
-                "state": "FAIL",
-                "fail_reason": "TRANSFER_RISK",
-                "transfer_bill_no": "wx_failed_reason_01",
-            },
-        },
+        wr.wxpay, "query_transfer",
+        lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
     )
-
-    response = admin_client.post(
-        "/admin/api/withdraws/adminfailedreason01/refresh",
+    r = admin_client.post(
+        "/admin/api/withdraws/reconcile", params={"older_than_minutes": 0},
         headers=_auth(finance_token),
     )
-
-    assert response.status_code == 200, response.text
-    assert response.json()["status"] == "failed"
-    assert response.json()["wechat_state"] == "FAIL"
-    assert response.json()["transfer_bill_no"] == "wx_failed_reason_01"
-    assert response.json()["fail_reason"] == (
-        "微信转账失败：该笔转账存在风险，已被微信拦截（TRANSFER_RISK）"
-    )
+    assert r.status_code == 200, r.text
+    assert "checked" in r.json() and "resolved" in r.json()

tests/test_compare_harvest.py

@@ -9,16 +9,17 @@ pricebot 用 httpx mock,不真连(同 test_compare_proxy)。
 """
 from __future__ import annotations
 
+import json
 import uuid
 from unittest.mock import MagicMock, patch
 
 import httpx
-from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
 from app.schemas.compare_record import ComparisonRecordIn
+from sqlalchemy import select
 
 
 def _tid() -> str:
@@ -215,9 +216,7 @@ def test_price_step_done_harvests_success(client) -> None:
                   "action": {"command": "done", "params": _done_params()},
                   "trace_url": "https://price.shaguabijia.com/traces/done2/"}
     p, _cap = _mock_pricebot(done_frame)
-    with p, patch(
-        "app.api.v1.compare.backfill_comparison_llm_cost"
-    ) as backfill:
+    with p:
         r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
     assert r.status_code == 200
     with SessionLocal() as db:
@@ -225,7 +224,6 @@ def test_price_step_done_harvests_success(client) -> None:
         assert rec is not None and rec.status == "success"
         assert rec.best_platform_id == "meituan"
         assert rec.saved_amount_cents == 500
-        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_trace_finalize_harvests_abort(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -1,161 +0,0 @@
-from __future__ import annotations
-
-from datetime import UTC, datetime, timedelta
-
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal
-from app.models.app_config import AppConfig
-from app.models.comparison import ComparisonRecord
-from app.services import comparison_llm_backfill
-
-
-def _record(
-    trace_id: str,
-    *,
-    status: str = "success",
-    created_at: datetime | None = None,
-) -> int:
-    with SessionLocal() as db:
-        rec = ComparisonRecord(
-            trace_id=trace_id,
-            status=status,
-            created_at=created_at or datetime.now(),
-        )
-        db.add(rec)
-        db.commit()
-        return rec.id
-
-
-def _delete(record_id: int) -> None:
-    with SessionLocal() as db:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is not None:
-            db.delete(rec)
-            db.commit()
-
-
-def test_backfill_retries_then_persists_cost(monkeypatch):
-    record_id = _record("llm-retry-1")
-    calls = [
-        {
-            "model": "unknown-model",
-            "error": None,
-            "usage": {"prompt_tokens": 1000, "completion_tokens": 500},
-        }
-    ]
-    responses = iter([[], calls])
-    monkeypatch.setattr(
-        comparison_llm_backfill,
-        "fetch_llm_calls",
-        lambda trace_id: next(responses),
-    )
-    sleeps: list[float] = []
-    monkeypatch.setattr(comparison_llm_backfill.time, "sleep", sleeps.append)
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
-
-    try:
-        assert comparison_llm_backfill.backfill_comparison_llm_cost(
-            record_id, "llm-retry-1", attempts=2, retry_delays=(0.25,)
-        )
-        assert sleeps == [0.25]
-        with SessionLocal() as db:
-            rec = db.get(ComparisonRecord, record_id)
-            assert rec.input_tokens == 1000
-            assert rec.output_tokens == 500
-            assert rec.llm_cost_yuan is not None
-            assert rec.llm_calls == calls
-    finally:
-        _delete(record_id)
-
-
-def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
-    missing_id = _record("llm-repair-missing")
-    running_id = _record("llm-repair-running", status="running")
-    calls = [
-        {
-            "model": "unknown-model",
-            "error": None,
-            "usage": {"prompt_tokens": 100, "completion_tokens": 20},
-        }
-    ]
-    seen: list[str] = []
-
-    def fetch(trace_id: str) -> list[dict]:
-        seen.append(trace_id)
-        return calls
-
-    monkeypatch.setattr(comparison_llm_backfill, "fetch_llm_calls", fetch)
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
-    try:
-        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
-            limit=10, lookback_days=1
-        )
-        assert result["repaired"] >= 1
-        assert "llm-repair-missing" in seen
-        assert "llm-repair-running" not in seen
-        with SessionLocal() as db:
-            assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
-            assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
-    finally:
-        _delete(missing_id)
-        _delete(running_id)
-
-
-def test_repair_excludes_records_before_current_price_config(monkeypatch):
-    price_changed_at = datetime.now(UTC) - timedelta(hours=1)
-    before_change = (price_changed_at - timedelta(minutes=30)).astimezone(CN_TZ)
-    after_change = (price_changed_at + timedelta(minutes=30)).astimezone(CN_TZ)
-    before_id = _record(
-        "llm-before-price-change",
-        created_at=before_change.replace(tzinfo=None),
-    )
-    after_id = _record(
-        "llm-after-price-change",
-        created_at=after_change.replace(tzinfo=None),
-    )
-    with SessionLocal() as db:
-        existing = db.get(AppConfig, "llm_token_price")
-        if existing is not None:
-            db.delete(existing)
-            db.flush()
-        db.add(
-            AppConfig(
-                key="llm_token_price",
-                value={"default": {"input_per_1m": 1, "output_per_1m": 1}},
-                updated_at=price_changed_at,
-            )
-        )
-        db.commit()
-
-    seen: list[str] = []
-
-    def backfill(record_id: int, trace_id: str, **kwargs) -> bool:
-        seen.append(trace_id)
-        return True
-
-    monkeypatch.setattr(
-        comparison_llm_backfill,
-        "backfill_comparison_llm_cost",
-        backfill,
-    )
-    try:
-        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
-            limit=10_000,
-            lookback_days=1,
-        )
-        assert "llm-after-price-change" in seen
-        assert "llm-before-price-change" not in seen
-        assert result["repaired"] == len(seen)
-        assert result["unresolved"] == 0
-    finally:
-        _delete(before_id)
-        _delete(after_id)
-        with SessionLocal() as db:
-            config = db.get(AppConfig, "llm_token_price")
-            if config is not None:
-                db.delete(config)
-                db.commit()

tests/test_comparison_result_in_skipped.py

@@ -1,17 +0,0 @@
-"""ComparisonResultIn 保留逐平台 skipped_dish_count(落库/回读用)。
-
-记录页三平台网格要按逐平台缺菜数标"缺少 X 个菜品"; pricebot 现把 skipped_dish_count 冗余进
-comparison_results 行。上报边界必须显式声明该字段, 否则 model_dump() 会静默丢弃(POST 路径),
-记录页拿不到。纯 schema, 不碰 DB。
-"""
-from app.schemas.compare_record import ComparisonResultIn
-
-
-def test_comparison_result_in_preserves_skipped_dish_count():
-    r = ComparisonResultIn(platform_id="jd_waimai", price=25.0, skipped_dish_count=2)
-    assert r.model_dump().get("skipped_dish_count") == 2
-
-
-def test_comparison_result_in_skipped_defaults_none():
-    r = ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True)
-    assert r.model_dump().get("skipped_dish_count") is None

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_invite_cash_withdraw.py

@@ -127,51 +127,6 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None:
         db.close()
 
 
-def test_invite_cash_reject_recreates_missing_account(client, monkeypatch) -> None:
-    """A legacy or hand-written order without CoinAccount can still be refunded."""
-    _patch_userinfo(monkeypatch, "openid_ic_missing_account")
-    token = _login(client, "13800004012")
-    _seed_balances(client, token, "13800004012", cash=0, invite_cash=200)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    response = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    assert response.status_code == 200, response.text
-    bill = response.json()["out_bill_no"]
-
-    with SessionLocal() as db:
-        user = db.execute(
-            select(User).where(User.phone == "13800004012")
-        ).scalar_one()
-        db.delete(db.get(CoinAccount, user.id))
-        db.commit()
-
-    _reject(bill, "missing account fixture")
-
-    with SessionLocal() as db:
-        user = db.execute(
-            select(User).where(User.phone == "13800004012")
-        ).scalar_one()
-        account = db.get(CoinAccount, user.id)
-        order = crud_wallet._get_withdraw_or_raise(db, bill)
-        refunds = db.execute(
-            select(InviteCashTransaction).where(
-                InviteCashTransaction.user_id == user.id,
-                InviteCashTransaction.biz_type == "invite_withdraw_refund",
-                InviteCashTransaction.ref_id == bill,
-            )
-        ).scalars().all()
-
-        assert account is not None
-        assert account.invite_cash_balance_cents == 200
-        assert order.status == "rejected"
-        assert order.fail_reason == "missing account fixture"
-        assert len(refunds) == 1 and refunds[0].amount_cents == 200
-
-
 def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     """两账户各提各的不串:提 invite_cash(拒绝→验证退款回原账户),再提 cash,各扣各账户互不串。"""
     _patch_userinfo(monkeypatch, "openid_ic_3")

tests/test_llm_cost.py

@@ -132,7 +132,6 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
     from app.models.app_config import AppConfig
     from app.models.comparison import ComparisonRecord
     from app.repositories import app_config
-    from app.services import comparison_llm_backfill
 
     sample = [
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
@@ -140,12 +139,7 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
     ]
-    monkeypatch.setattr(
-        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
-    )
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
+    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
 
     db = SessionLocal()
     try:

tests/test_pricebot_llm_calls.py

@@ -1,55 +0,0 @@
-from unittest.mock import MagicMock
-
-from app.core.config import settings
-from app.services import pricebot_llm_calls
-
-
-def test_auth_probe_accepts_matching_secret(monkeypatch):
-    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
-    response = MagicMock(status_code=200)
-    get = MagicMock(return_value=response)
-    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
-
-    assert pricebot_llm_calls.pricebot_llm_auth_ready() is True
-    assert get.call_count == len(settings.pricebot_instances)
-    assert all(
-        call.kwargs["headers"]["X-Internal-Secret"] == "same-secret"
-        for call in get.call_args_list
-    )
-
-
-def test_auth_probe_rejects_mismatched_secret(monkeypatch):
-    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "app-server-secret")
-    monkeypatch.setattr(
-        pricebot_llm_calls.httpx,
-        "get",
-        MagicMock(return_value=MagicMock(status_code=403)),
-    )
-
-    assert pricebot_llm_calls.pricebot_llm_auth_ready() is False
-
-
-def test_fetch_falls_back_to_other_instance_when_hash_target_is_empty(monkeypatch):
-    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
-    monkeypatch.setattr(
-        settings,
-        "PRICEBOT_INSTANCES",
-        "http://pricebot-1:8000,http://pricebot-2:8000",
-    )
-    monkeypatch.setattr(
-        pricebot_llm_calls,
-        "pick_pricebot",
-        lambda trace_id: "http://pricebot-1:8000",
-    )
-    calls = [{"model": "qwen", "usage": {"prompt_tokens": 1}}]
-
-    def get(url, **kwargs):
-        response = MagicMock(status_code=200)
-        response.json.return_value = {
-            "calls": [] if "pricebot-1" in url else calls
-        }
-        return response
-
-    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
-
-    assert pricebot_llm_calls.fetch_llm_calls("trace-after-rescale") == calls

tests/test_withdraw.py

@@ -9,7 +9,7 @@ from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.user import User
-from app.models.wallet import CoinAccount, WechatTransferAuthorization, WithdrawOrder
+from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
 from app.repositories import wallet as crud_wallet
 
 
@@ -232,78 +232,6 @@ def test_withdraw_approve_transfer_fail_refunds(client, monkeypatch) -> None:
     assert r.json()["items"][0]["status"] == "failed"
 
 
-def test_approved_withdraw_query_fail_exposes_wechat_reason(client, monkeypatch) -> None:
-    """审核通过后微信终态失败时，保存官方失败码对应原因并退回余额。"""
-    monkeypatch.setattr(
-        "app.integrations.wxpay.code_to_userinfo",
-        lambda code: {
-            "openid": "openid_query_fail",
-            "nickname": None,
-            "avatar_url": None,
-            "raw": {},
-        },
-    )
-    monkeypatch.setattr(
-        "app.integrations.wxpay.create_transfer",
-        lambda openid, amount_fen, out_bill_no, user_name=None: {
-            "status_code": 200,
-            "data": {
-                "state": "WAIT_USER_CONFIRM",
-                "package_info": "pkg_query_fail",
-                "transfer_bill_no": "tb_query_fail",
-            },
-        },
-    )
-    token = _login(client, "13800002025")
-    _seed_cash(client, token, "13800002025", 100)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    response = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 50},
-        headers=_auth(token),
-    )
-    bill = response.json()["out_bill_no"]
-    _approve(bill)
-
-    monkeypatch.setattr(
-        "app.integrations.wxpay.query_transfer",
-        lambda out_bill_no: {
-            "status_code": 200,
-            "data": {
-                "state": "FAIL",
-                "fail_reason": "PAYEE_ACCOUNT_ABNORMAL",
-                "transfer_bill_no": "tb_query_fail",
-            },
-        },
-    )
-    response = client.get(
-        "/api/v1/wallet/withdraw/status",
-        params={"out_bill_no": bill},
-        headers=_auth(token),
-    )
-
-    assert response.status_code == 200, response.text
-    assert response.json()["status"] == "failed"
-    assert response.json()["wechat_state"] == "FAIL"
-    assert response.json()["fail_reason"] == (
-        "微信转账失败：用户微信账户收款异常（PAYEE_ACCOUNT_ABNORMAL）"
-    )
-    with SessionLocal() as db:
-        order = db.execute(
-            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill)
-        ).scalar_one()
-        assert order.transfer_bill_no == "tb_query_fail"
-    account = client.get("/api/v1/wallet/account", headers=_auth(token))
-    assert account.json()["cash_balance_cents"] == 100
-
-
-def test_unknown_wechat_fail_reason_code_is_preserved() -> None:
-    assert crud_wallet._wechat_terminal_failure_reason(
-        {"fail_reason": "NEW_WECHAT_REASON"}, "FAIL"
-    ) == "微信转账失败：NEW_WECHAT_REASON"
-
-
 def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
     """#2 同 out_bill_no 重试:只转一次,第二次返回同一单,余额只扣一次。"""
     monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_idem", "nickname": None, "avatar_url": None, "raw": {}})

tests/test_withdraw_ledger_check.py

@@ -0,0 +1,151 @@
+"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。
+
+历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对,
+而 source=invite_cash 的提现单流水其实在 invite_cash_transaction 表,导致每笔邀请提现单都被
+误报「缺扣款/缺退款流水」。这里用真实提现 API 造单 + before/after 差值断言锁定修复:
+  1) 邀请提现单不再污染普通现金账的缺流水计数;
+  2) 邀请账户已被纳入对账(能抓到它自己的缺流水);
+  3) 普通现金账的原有对账未被改坏。
+
+conftest 的库是 session 级共享、测试间不清,故一律用 before/after 差值,只反映本用例造的数据。
+"""
+from __future__ import annotations
+
+from sqlalchemy import delete, select
+
+from app.admin.repositories.queries import withdraw_ledger_check
+from app.db.session import SessionLocal
+from app.models.user import User
+from app.models.wallet import CoinAccount, InviteCashTransaction
+from app.repositories import wallet as crud_wallet
+
+
+def _login(client, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _patch_userinfo(monkeypatch, openid: str) -> None:
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}},
+    )
+
+
+def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None:
+    client.get("/api/v1/wallet/account", headers=_auth(token))
+    db = SessionLocal()
+    try:
+        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
+        acc = db.get(CoinAccount, user.id)
+        acc.cash_balance_cents = cash
+        acc.invite_cash_balance_cents = invite_cash
+        db.commit()
+    finally:
+        db.close()
+
+
+def _reject(bill: str, reason: str = "测试拒绝") -> None:
+    db = SessionLocal()
+    try:
+        crud_wallet.reject_withdraw(db, bill, reason)
+    finally:
+        db.close()
+
+
+def _ledger() -> dict:
+    db = SessionLocal()
+    try:
+        return withdraw_ledger_check(db)
+    finally:
+        db.close()
+
+
+def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None:
+    """核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction,
+    不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)。"""
+    before = _ledger()
+
+    _patch_userinfo(monkeypatch, "openid_lc_1")
+    token = _login(client, "13800005001")
+    _seed_balances(client, token, "13800005001", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # rejected + 退款流水落 invite_cash_transaction
+
+    after = _ledger()
+
+    # 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报)
+    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
+    assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"]
+    # 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺
+    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]
+    assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"]
+
+
+def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None:
+    """删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False,
+    证明邀请账户已真正纳入对账(修复前邀请账完全不校验、永远报不出问题)。"""
+    _patch_userinfo(monkeypatch, "openid_lc_2")
+    token = _login(client, "13800005002")
+    _seed_balances(client, token, "13800005002", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    bill = r.json()["out_bill_no"]
+
+    before = _ledger()
+    db = SessionLocal()
+    try:
+        db.execute(
+            delete(InviteCashTransaction).where(
+                InviteCashTransaction.ref_id == bill,
+                InviteCashTransaction.biz_type == "invite_withdraw",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+    after = _ledger()
+
+    assert (
+        after["invite_missing_withdraw_txn_count"]
+        == before["invite_missing_withdraw_txn_count"] + 1
+    )
+    assert after["ok"] is False
+
+
+def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
+    """普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。"""
+    _patch_userinfo(monkeypatch, "openid_lc_3")
+    token = _login(client, "13800005003")
+    _seed_balances(client, token, "13800005003", cash=500, invite_cash=0)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    before = _ledger()
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    after = _ledger()
+
+    # 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响
+    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
+    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]