修复:ComparisonResultIn 补 skipped_dish_count(逐平台缺菜数落库)

pricebot 把逐平台缺菜数冗余进 comparison_results 每行,但上报入参 schema ComparisonResultIn 未声明该字段,model_dump() 会在 POST 路径静默丢弃, 记录页三平台网格拿不到逐格「缺少 X 个菜品」。显式声明补齐 + 回归测试。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
修复比价记录平均 TOKEN 成本采集与回填 (#182 )
2026-07-27 16:01:50 +08:00 · 2026-07-27 15:51:19 +08:00 · 2026-07-27 13:33:55 +08:00 · 2026-07-27 10:35:15 +08:00 · 2026-07-27 10:24:32 +08:00 · 2026-07-27 10:15:54 +08:00
39 changed files with 1061 additions and 612 deletions
@@ -139,6 +139,11 @@ 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 之类
@@ -57,12 +57,11 @@ def _reward_video_rows(
        stmt = stmt.where(AdRewardRecord.user_id == user_id)

    records = list(db.execute(stmt).scalars())
-    # S2S 发奖回调本身不携带实际填充的 ADN/底层 rit；按客户端在展示时上报的
-    # ad_session_id 回填。这样“纯发奖”行也能在运营后台追溯到真实广告网络。
-    session_ids = {rec.ad_session_id for rec in records if rec.ad_session_id}
+    # S2S 发奖回调不携带实际填充 ADN；用相同用户和 ad_session_id 的展示记录回填。
+    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
    impression_by_session = {
-        (rec.user_id, rec.ad_session_id): rec
-        for rec in db.execute(
+        (record.user_id, record.ad_session_id): record
+        for record in db.execute(
            select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
        ).scalars()
    } if session_ids else {}
@@ -172,7 +171,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:
@@ -185,14 +184,11 @@ 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]]]:
-    """构建信息流来源回填索引。
-
-    新客户端会把 ADN 直接随 feed-reward 上报；旧记录可能缺失。展示收益记录的
-    ``ad_session_id`` 是每条 impressionId，而发奖记录保留的是整场会话 ID，因此先按
-    会话精确匹配；匹配不到时仅允许按 ``user + trace_id + 原始 eCPM`` 回填，且候选 ADN
-    必须唯一。trace 内存在多个网络时保持空值，避免错误归因。
-    """
+) -> tuple[
+    dict[tuple[int, str], tuple[str | None, str | None]],
+    dict[tuple[int, str, str], tuple[str | None, str | None]],
+]:
+    """为旧信息流发奖记录构建安全来源索引。"""
    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:
@@ -227,7 +223,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
@@ -22,7 +22,7 @@ report_date / reward_date 归日。
 """
 from __future__ import annotations

-from datetime import UTC, datetime, time, timedelta
+from datetime import UTC, datetime, 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"}

-# GroMore 官方说明第三方 ADN 的 Reporting API 最晚约 13:50 更新。只有 D+1 14:00
-# 之后完成的同步才标记为「API 同步窗口完成」；这不代表覆盖全部 ADN 或最终结算。
-_PANGLE_API_FINAL_SYNC_TIME = time(hour=14)
+# 激励视频未满足有效播放条件时不计客户端预估收益。客户端仍会在 onAdShow
+# 上报 eCPM，随后才在关闭时补报以下终态，因此必须在展示/发奖合并后修正收益。
+_ZERO_REVENUE_REWARD_VIDEO_STATUSES = frozenset({"closed_early", "too_short"})


 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
@@ -96,34 +96,13 @@ _REWARD_DETAIL_KEYS = (

 def _reward_detail(row: dict) -> dict:
    """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
-    detail = {k: row[k] for k in _REWARD_DETAIL_KEYS}
-    # 发奖明细必须保留自己的广告网络，不能复用整场聚合父行的来源：
-    # 同一次比价/领券可能先后由不同 ADN 填充。
+    detail = {key: row[key] for key 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,
    *,
@@ -211,10 +190,12 @@ def ad_revenue_report(
            "has_impression": True,
            "impressions": 1,
            "ecpm": rec.ecpm_raw,
-            # 客户端 SDK 展示预估收益(元)= 后端留存 getEcpm 元/千次 ÷ 1000。
-            # 这里不能复用发奖防作弊的 ¥500 CPM 钳顶：钳顶只限制金币成本，不改变广告已产生的
-            # 收入估值。onAdShow 已发生即计展示收入，是否看满只影响发奖，不影响广告收入。
-            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0, 6),
+            # 单次展示收益(元)= 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,
+            ),
            "adn": rec.adn,
            "slot_id": rec.slot_id,
            "sub_rewards": [],
@@ -229,6 +210,11 @@ 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({
@@ -289,10 +275,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(
-            rewards.parse_ecpm_yuan(g["ecpm"]) / 1000.0
+            min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
            for g in group if g.get("ecpm")
        ), 6)
        events.append({
@@ -382,15 +368,13 @@ def ad_revenue_report(
        for d in sorted(daily_map.values(), key=lambda x: x["date"])
    ]

-    # GroMore 排序价预估 / ADN Reporting API 收益(T+1 入库):汇总 + 按天趋势级展示,
+    # 穿山甲后台收益(GroMore 数据 API,T+1 入库 ad_pangle_daily_revenue):汇总 + 按天趋势级展示,
    # 与上面客户端自报 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,
@@ -408,12 +392,6 @@ 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。
@@ -438,50 +416,41 @@ def ad_revenue_report(
            for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
        ]

-    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()
-        }
+    # 分广告类型小计(按 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()
+    }

-    # 原始 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
-        )
-    )
-
-    # 分场景小计，同 type_stats 基于全量 events，供数据大盘「领券广告 / 比价广告」卡使用。
-    # feed_scene 为空的激励视频 / 历史数据不计入任何场景桶。
-    scene_stats = _aggregate_stats(lambda e: e.get("feed_scene"))
+    # 分场景小计(按 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()
+    }

    # DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
    # 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
@@ -506,11 +475,9 @@ def ad_revenue_report(
        "truncated": len(main_rows) > offset + limit,
        "total_impressions": total_impressions,
        "total_revenue_yuan": total_revenue_yuan,
-        # GroMore 排序价预估 + ADN Reporting API 收益;非全量视图或无数据为 None。
+        # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 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,
@@ -518,7 +485,6 @@ def ad_revenue_report(
        "daily": daily,
        "hourly": hourly,
        "type_stats": type_stats,
-        "category_stats": category_stats,
        "scene_stats": scene_stats,
        "dau": dau,
        "items": main_rows[offset:offset + limit],
@@ -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, asc, case, desc, func, or_, select
+from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session

 from app.core import rewards
@@ -45,6 +45,72 @@ _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_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]:
@@ -817,6 +883,7 @@ 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


@@ -1228,6 +1295,11 @@ 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,
@@ -1319,7 +1391,10 @@ def user_coin_records(
            "coin": rec.amount,
        })

-    rows.sort(key=lambda r: r["created_at"], reverse=True)
+    # 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)
    has_more = len(rows) > offset + limit

    # 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)
@@ -99,7 +99,6 @@ 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"],
@@ -108,8 +107,6 @@ 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"],
@@ -86,7 +86,7 @@ def get_user_reward_stats(
    date_to: Annotated[datetime | None, Query()] = None,
 ) -> UserRewardStats:
    """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
-    if user_repo.get_user_by_id(db, user_id) is None:
+    if not user_repo.user_exists(db, user_id):
        raise HTTPException(status_code=404, detail="用户不存在")
    return UserRewardStats(
        **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)
@@ -49,12 +49,12 @@ class AdRevenueDaily(BaseModel):

    date: str = Field(..., description="北京时间 YYYY-MM-DD")
    impressions: int = Field(..., description="当天展示条数合计")
-    revenue_yuan: float = Field(..., description="当天客户端 SDK 展示预估合计(元;后端留存 eCPM 折算)")
+    revenue_yuan: float = Field(..., description="当天客户端有效预估收益合计(元;eCPM 折算)")
    pangle_revenue_yuan: float | None = Field(
-        None, description="当天 GroMore 排序价预估(元;revenue,非结算收入);非全量视图/无数据为空"
+        None, description="当天穿山甲后台预估收益(元;GroMore revenue);非全量视图/无数据为空"
    )
    pangle_api_revenue_yuan: float | None = Field(
-        None, description="当天 ADN Reporting API 收益(元;GroMore api_revenue);未配/当天/无数据为空"
+        None, description="当天穿山甲收益Api(元;GroMore api_revenue,更接近结算);未配/当天/无数据为空"
    )
    expected_coin: int = Field(..., description="当天应发金币合计")
    actual_coin: int = Field(..., description="当天实发金币合计")
@@ -71,11 +71,10 @@ class AdRevenueHourly(BaseModel):


 class AdRevenueTypeStat(BaseModel):
-    """展示条数、SDK 展示预估收益与按展示次数加权的 SDK eCPM。"""
+    """按广告类型(ad_type)的小计:展示条数 + 预估收益(eCPM 由前端用 收益÷展示×1000 算)。"""

    impressions: int = Field(..., description="该类型展示条数合计")
    revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
-    ecpm_yuan: float = Field(..., description="按展示次数加权的 SDK eCPM(元/千次)")


 class AdRevenueRow(BaseModel):
@@ -101,15 +100,15 @@ class AdRevenueRow(BaseModel):
    ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
    revenue_yuan: float = Field(
        ...,
-        description="本次 SDK 展示预估收益(元)=后端留存 eCPM 元 ÷ 1000;是否满足发奖条件不改变展示收入预估",
+        description="本次有效展示预估收益(元)= eCPM元 ÷ 1000;纯发奖、激励视频提前关闭/时长不足=0",
    )
    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,11 +142,7 @@ class AdRevenueReportOut(BaseModel):
    )
    type_stats: dict[str, AdRevenueTypeStat] = Field(
        default_factory=dict,
-        description="按原始广告类型(ad_type)小计，供筛选与排查使用",
-    )
-    category_stats: dict[str, AdRevenueTypeStat] = Field(
-        default_factory=dict,
-        description="按经营分类小计：draw=draw+历史 feed，video=reward_video+withdrawal_video",
+        description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
    )
    scene_stats: dict[str, AdRevenueTypeStat] = Field(
        default_factory=dict,
@@ -163,28 +158,20 @@ 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="全量客户端 SDK 展示预估合计(元;后端留存 eCPM 折算)")
+    total_revenue_yuan: float = Field(..., description="全量客户端有效预估收益合计(元;eCPM 折算)")
    total_pangle_revenue_yuan: float | None = Field(
        None,
-        description="全量 GroMore 排序价预估合计(元;revenue,非结算收入)。GroMore 无用户/类型/场景维度,"
+        description="全量穿山甲后台预估收益合计(元;GroMore revenue)。穿山甲无用户/类型/场景维度,"
        "仅「全量视图」(未按 user_id/ad_type/feed_scene 过滤)时有值,否则为 null",
    )
    total_pangle_api_revenue_yuan: float | None = Field(
        None,
-        description="全量 ADN Reporting API 收益合计(元;GroMore api_revenue,仅已配置回传的 ADN);"
+        description="全量穿山甲收益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="本次结果是否带 GroMore/ADN 收益(=全量视图且已同步到数据)。false 时前端显示「-」",
+        description="本次结果是否带穿山甲后台收益(=全量视图且已同步到数据)。false 时前端「穿山甲收益」显示「-」",
    )
    total_expected_coin: int = Field(..., description="全量应发金币合计")
    total_actual_coin: int = Field(..., description="全量实发金币合计")
@@ -32,7 +32,10 @@ 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
@@ -289,8 +289,7 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
    """客户端在广告展示后(onAdShow 读 getShowEcpm)上报 eCPM,落库做内部收益统计/对账。

    Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
-    丢一两条不影响发奖业务(收入另由 ADN Reporting API 对账)。eCPM 与发奖(S2S)是两条独立流,
-    不逐条关联。
+    丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
    """
    attributed_trace_id = crud_ecpm.attributable_trace_id(
        db,
@@ -25,7 +25,7 @@ import uuid
 from typing import Any

 import httpx
-from fastapi import APIRouter, HTTPException, Request, status
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool

 from app.api.deps import DbSession, OptionalUser
@@ -36,6 +36,7 @@ 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")

@@ -80,7 +81,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,
-) -> None:
+) -> int:
    with SessionLocal() as db:
        rec, newly_success = crud_compare.harvest_done(
            db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -102,6 +103,7 @@ def _harvest_done_blocking(
        # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
        # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
        # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
+        return rec.id


 def _harvest_abort_blocking(
@@ -243,7 +245,12 @@ async def intent_precoupon_step(


@router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
+async def price_step(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    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 更新成终态。
@@ -252,12 +259,15 @@ async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dic
    if action.get("command") == "done" and not resp.get("continue", True):
        done_params = action.get("params") or {}
        try:
-            await run_in_threadpool(
+            record_id = 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
@@ -16,8 +16,6 @@ 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 (
@@ -30,8 +28,7 @@ from app.schemas.compare_record import (
    ComparisonRecordOut,
    ComparisonRecordPage,
 )
-from app.services.llm_cost import compute_llm_cost, get_llm_prices
-from app.services.pricebot_llm_calls import fetch_llm_calls
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost

 logger = logging.getLogger("shagua.compare_record")

@@ -121,32 +118,7 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
    """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
    独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    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()
+    backfill_comparison_llm_cost(record_id, trace_id)


@router.get(
@@ -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 ADN Reporting 收益),不参与发奖。
+    # 仅用于按天拉 GroMore 收益报表(revenue 预估收益 + api_revenue 收益Api),不参与发奖。
    # 该 API 只能查【GroMore 聚合代码位】的数据(=我们 useMediation 的口径),非穿山甲 SDK 数据;
    # 且不提供用户/设备维度(官方明确),故收益只能落到 日期×代码位 汇总,不能挂到逐条事件。
    # 子账号(role_id≠user_id)需主账号在「角色管理」授予「查看全部数据」权限,否则查不到
@@ -377,6 +377,12 @@ 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。
@@ -0,0 +1,63 @@
+"""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
@@ -14,8 +14,8 @@
  - 只返回【GroMore 聚合代码位】在 GroMore 内的数据(=我们 useMediation 的口径),
    查不到穿山甲 SDK 自身的数据;
  - **不提供分用户/设备维度**(官方 FAQ 明确拒绝),最细到 日期×应用×代码位×广告源;
-  - `revenue` = 排序价/竞价实时价预估(元,非结算收入);`api_revenue` = 各 ADN 经 Reporting
-    回传、按实时汇率折算账号币种的收益,需后台为该 ADN 配置 Reporting 才有、且不支持当天;
+  - `revenue` = 预估收益(元,所有 ADN 都有);`api_revenue` = 收益Api(各 ADN 经 Reporting
+    回传、按实时汇率折算账号币种,更接近结算),需后台为该 ADN 配置 Reporting 才有、且不支持当天;
  - 「今天」与「今天以前」必须分开查;天级跨度 ≤ 1 个月、不早于 12 个月。
 """
 from __future__ import annotations
@@ -60,6 +60,10 @@ 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 (
@@ -103,6 +107,7 @@ 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:
@@ -112,6 +117,7 @@ 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")
@@ -1,15 +1,15 @@
-"""GroMore 天级排序价预估与 ADN Reporting 收益(定时拉取入库)。
+"""穿山甲 GroMore 天级收益报表(后台结算口径,定时拉取入库)。

 每行 = GroMore 数据 API 返回的一条「日期 × 应用 × 代码位」聚合收益(`integrations/pangle_report`
-+ `scripts/sync_pangle_revenue` 落库)，与 `ad_ecpm_record`(客户端 SDK eCPM 折算的预估)
-互为对照:
+ `scripts/sync_pangle_revenue` 落库)。**权威/预估收益的来源**,与 `ad_ecpm_record`(客户端自报
+eCPM 折算的预估)互为对照:

- `revenue_yuan` ← 接口 `revenue`(排序价/竞价实时价预估,元,不是结算收入);
- `api_revenue_yuan` ← 接口 `api_revenue`(各 ADN Reporting 回传收益,元,更接近结算;
+- `revenue_yuan` ← 接口 `revenue`(预估收益,元;排序价×展示/1000,所有 ADN 都有);
+- `api_revenue_yuan` ← 接口 `api_revenue`(收益Api,元;各 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)
-    # ADN Reporting API 收益(元)← api_revenue;未配 Reporting / 当天等情况不返回 → NULL。
+    # 收益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)
@@ -1,8 +1,8 @@
 """广告 eCPM 上报 CRUD(内部收益统计/对账)。

 客户端在广告展示后(onAdShow)读到 eCPM,经鉴权接口上报,这里落库。鉴权接口已确保
-user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响发奖业务;
-汇总收入以 ADN Reporting API 和最终结算单为准。
+user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响业务,
+穿山甲后台报表是结算权威兜底。
 """
 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 —— 丢一条不影响业务
-        # (收入另由 ADN Reporting API 对账),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
+        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
        # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
        # 旧逻辑在此 raise 成 500(本应静默吞掉)。
        existing = _find_by_session_global(db, ad_session_id)
@@ -1,13 +1,12 @@
 """穿山甲 GroMore 天级收益 读写(`ad_pangle_daily_revenue` 表)。

 `scripts/sync_pangle_revenue` 拉数后调 `upsert_daily_rows` 落库(同一(日期×应用×代码位×广告源)
-幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取 GroMore/ADN 收益做
+幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取「穿山甲后台收益」做
 汇总/趋势级展示。穿山甲无用户维度,故这里不涉及 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
@@ -24,7 +23,6 @@ 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]:
@@ -89,7 +87,6 @@ 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,
@@ -106,12 +103,11 @@ 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, synced_at in db.execute(stmt).all():
+    for report_date, rev, api_rev, imp 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
@@ -55,13 +55,22 @@ 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 {}

-    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
-    best = None
+    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]
-    if priced:
+    clean = [r for r in priced if not _is_short(r)]
+    best = None
+    if clean:
        best = min(
-            priced,
+            clean,
            key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
        )

@@ -196,14 +205,29 @@ def upsert_record(
 # ============================================================


-def _derive_from_results(results: list[dict]) -> dict:
+def _derive_from_results(
+    results: list[dict], platform_results: dict | None = None
+) -> dict:
    """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
+
+    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
+    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
+    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
+    _pr = platform_results or {}
+
+    def _is_short(r: dict) -> bool:
+        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
+        pid = r.get("platform_id")
+        info = _pr.get(pid) if pid else None
+        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
+
    priced = [r for r in results if r.get("price") is not None]
+    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
    best = None
-    if priced:
+    if clean:
        best = min(
-            priced,
+            clean,
            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)
@@ -389,7 +413,7 @@ def harvest_done(
    返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
    行不存在(理论上帧0已建;防御)则新建。"""
    results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results)
+    derived = _derive_from_results(results, done_params.get("platform_results"))
    # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
    items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
    fields = dict(
@@ -93,6 +93,11 @@ 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()
@@ -85,6 +85,10 @@ 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):
@@ -0,0 +1,156 @@
+"""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,
+    }
@@ -20,19 +20,67 @@ 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 []
-    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)
+    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)
    return []
@@ -0,0 +1,40 @@
+# 比价 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 仍保留时才能准确回填；
+原始调用已经清理的记录不能用估算值冒充真实成本。
@@ -1,13 +1,13 @@
 # 穿山甲 GroMore 收益拉取 定时任务 — 运维手册

-> 对象:维护「每天拉 GroMore / ADN 收益入库」这套定时任务的同事。
+> 对象:维护「每天拉穿山甲后台收益入库」这套定时任务的同事。
 > 🔒 服务器登录信息见**私密交接清单**,不入库。

 ## 它是什么
-admin「广告收益报表」里的 GroMore / ADN 收益读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查询**。GroMore 的 T+1 初值约 10:00 可用，但第三方 ADN Reporting 数据可能到 13:50 才更新，所以需要早晚各拉一次。
+admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查穿山甲**。穿山甲只通过 GroMore 数据 API 给数、且 **T+1**(次日约 10:00 出昨天的数),所以每天得拉一次入库,报表才会往前走。

- 每天 10:30 拉初值、14:30 拉日终值，均由 `scripts/sync_pangle_revenue.py` 以 `--days 3` 回补近 3 天。
- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(排序价预估)+ `api_revenue`(ADN Reporting 回传，更接近结算)。
+- 每天 10:30 跑一轮 `scripts/sync_pangle_revenue.py`,默认 `--days 3` 回补近 3 天。
+- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(预估)+ `api_revenue`(结算口径)。
 - **幂等 upsert**:同一(日期×应用×代码位)重跑只覆盖、不重复,故回补 / 重跑 / catch-up 都安全。
 - 穿山甲无用户/设备维度 → 只能落「汇总/趋势级」,报表带 user_id 过滤时这块收益置空(显示「-」)。

@@ -30,15 +30,15 @@ admin「广告收益报表」里的 GroMore / ADN 收益读的是**本地表 `ad
 ```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 或 14:30)
+systemctl list-timers pangle-revenue.timer   # 确认下次触发时间(应是次日 10:30)
 ```

 ## 怎么看健康 / 手动跑一次
 ```bash
-journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 收益合计
-sudo systemctl start pangle-revenue.service     # 立即手动跑一轮
+sudo systemctl start pangle-revenue.service     # 立即手动跑一轮(不等 10:30)
+journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 预估收益合计
 ```
-成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;排序价预估合计 ¥19.42`。
+成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;预估收益合计 ¥19.42`。
 > 看不到收益、提示 `PANGLE_REPORT_* 未配置`→ 回「上线前置」补 `.env`;报 118 → 子账号没授「查看全部数据」。

 ## 本机 Windows 开发(无 systemd)
@@ -57,11 +57,11 @@ sudo systemctl start pangle-revenue.service     # 立即手动跑一轮
 - `--start / --end`:指定闭区间(跨度 ≤ 31 天,接口上限 1 个月,超了报 114)。

 ## 注意事项
- **触发时间**:10:30 提供初值，14:30 覆盖为日终值；报表只把 D+1 14:00 后同步的数据标记为日终。
+- **触发时间**:`OnCalendar=*-*-* 10:30:00`。穿山甲 ~10:00 出数,故别早于 10: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；`revenue` 只是排序价估算，不能当结算收入。
+- **`api_revenue` 很稀疏**:测试应用 ADN 没配 Reporting → 全 0,仅 prod 个别位有;`revenue`(预估)才是稳的主力。
 - **DB 无关**:sqlite / postgres 均可(upsert 逐行 select-then-write,不像美团 ETL 需要 PG)。
 - **别和别的触发方式双跑**:本 systemd timer 与「手动 cron / 进程内任务」二选一,虽幂等不会重复入库,纯属多余。
 - **改脚本 / 改部署**:走 git + PR,由有 root 的人部署。
@@ -1,5 +1,5 @@
-# 拉 GroMore T+1 天级收益入库 —— 单轮跑,由 timer 每天 10:30、14:30 触发。
-# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的 GroMore/ADN 对账区块。
+# 每天拉穿山甲 GroMore T+1 天级收益入库 —— 单轮跑,由 pangle-revenue.timer 每天 10:30 触发。
+# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的「穿山甲后台收益(T+1)」区块。
 #
 # 仅用于 Linux 服务器;本机 Windows 开发无 systemd,直接手动跑脚本即可:
 #   .venv\Scripts\python -m scripts.sync_pangle_revenue            # 拉昨天(北京时间)
@@ -1,12 +1,11 @@
-# 每天 10:30 首次拉取、14:30 终值复拉 GroMore T+1 收益(Linux 服务器用)。
+# 每天 10:30 触发一次穿山甲 GroMore T+1 收益拉取入库(Linux 服务器用)。
 # 见 pangle-revenue.service 顶部注释的部署步骤。
 [Unit]
-Description=Run Pangle GroMore daily revenue sync at 10:30 and 14:30
+Description=Run Pangle GroMore daily revenue sync at 10:30

 [Timer]
-# 10:30 尽早展示初值；第三方 ADN Reporting 最晚约 13:50 更新，14:30 再拉一次作为日终值。
+# 穿山甲 T+1、次日约 10:00 出数;10:30 触发留 30min 余量。要错开整点扎堆可微调到 10:35。
 OnCalendar=*-*-* 10:30:00
-OnCalendar=*-*-* 14:30:00
 # 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日);叠加 --days 3 回补,漏一两天能自愈。
 Persistent=true
 AccuracySec=1min
@@ -1,211 +0,0 @@
-"""生成本地 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"
-    )
@@ -175,22 +175,22 @@ def seed(db) -> list[Feedback]:
        # 普通反馈 · 新端(带环境快照)· 无图
        fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
           source="profile",
-           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
+           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS 14", 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", android_version="13",
+           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
           created=ago(minutes=22)),
        # 比价反馈 · scene=找错商品 · 新端 · 无图
        fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
           source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
+           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", 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", android_version="14",
+           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI 14", 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", android_version="13",
+           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS 13", 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", android_version="13",
+           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
           created=ago(days=3)),
    ]
    db.add_all(feedbacks)
@@ -1,7 +1,7 @@
-"""每日拉取 GroMore 排序价预估与 ADN Reporting 收益入库(供 admin 广告收益对账)。
+"""每日拉取穿山甲 GroMore 天级收益报表入库(供 admin 广告收益报表的「穿山甲后台收益」)。

-GroMore 数据 API 为 T+1:次日约 10:00 出初值，第三方 ADN Reporting 最晚约 13:50 更新。
-线上由 systemd timer 在 10:30、14:30 各跑一次;历史数据可能订正,故支持回补近 N 天。
+GroMore 数据 API 为 T+1:次日穿山甲约 10:00 出数。建议线上每天 ~10:30 由 systemd timer 跑一次
+(默认拉【昨天】);穿山甲对历史数据可能订正,故支持回补近 N 天(幂等 upsert,重跑无害)。

 用法:
    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:
@@ -15,8 +15,6 @@ 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:
@@ -53,22 +51,18 @@ 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()
@@ -93,8 +87,6 @@ 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,
@@ -140,7 +132,7 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
        db.close()


-def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap() -> None:
+def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
    db = SessionLocal()
    phone = "18800009992"
    sessions = {
@@ -156,17 +148,13 @@ def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap

        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=ecpm_raw,
+                ecpm_raw="10000",
                report_date=PLAYBACK_DATE,
                created_at=created_at,
            ))
@@ -179,7 +167,7 @@ def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap
                ad_session_id=session_id,
                app_env="prod",
                our_code_id="prod-reward",
-                ecpm_raw=ecpm_raw,
+                ecpm_raw="10000",
                reward_date=PLAYBACK_DATE,
                created_at=created_at,
            ))
@@ -198,33 +186,22 @@ def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap

        revenue_by_status = {row["status"]: row["revenue_yuan"] for row in result["items"]}
        assert revenue_by_status == {
-            "closed_early": 0.1,
-            "too_short": 0.1,
-            "capped": 1.0,
+            "closed_early": 0.0,
+            "too_short": 0.0,
+            "capped": 0.1,
            "granted": 0.1,
        }
        assert result["total_impressions"] == 4
-        assert result["total_revenue_yuan"] == 1.3
+        assert result["total_revenue_yuan"] == 0.2
        assert len(result["daily"]) == 1
        assert result["daily"][0]["date"] == PLAYBACK_DATE
        assert result["daily"][0]["impressions"] == 4
-        assert result["daily"][0]["revenue_yuan"] == 1.3
-        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 1.3
+        assert result["daily"][0]["revenue_yuan"] == 0.2
+        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 0.2
        assert result["type_stats"]["reward_video"] == {
            "impressions": 4,
-            "revenue_yuan": 1.3,
-            "ecpm_yuan": 325.0,
+            "revenue_yuan": 0.2,
        }
-        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))
@@ -234,133 +211,54 @@ def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap
        db.close()


-def test_category_stats_merge_legacy_feed_and_withdrawal_video() -> None:
-    db = SessionLocal()
-    phone = "18800009993"
-    category_date = "2040-02-05"
-    try:
-        user = User(phone=phone, username="29999999993", register_channel="sms")
-        db.add(user)
-        db.flush()
-        db.add_all([
-            # Draw 经营分类必须包含新 draw 与历史 feed。
-            AdEcpmRecord(
-                user_id=user.id, ad_type="draw", ad_session_id="category-draw",
-                app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
-                report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
-            ),
-            AdEcpmRecord(
-                user_id=user.id, ad_type="feed", ad_session_id="category-feed",
-                app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
-                report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
-            ),
-            # 看视频经营分类必须包含福利与提现两个视频入口。
-            AdEcpmRecord(
-                user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
-                app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
-                report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
-            ),
-            AdEcpmRecord(
-                user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
-                app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
-                report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
-            ),
-        ])
-        db.commit()
-
-        result = ad_revenue.ad_revenue_report(
-            db,
-            date_from=category_date,
-            date_to=category_date,
-            user_id=user.id,
-            app_env="prod",
-            revenue_scope="all",
-        )
-
-        assert result["category_stats"] == {
-            "draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
-            "video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
-        }
-    finally:
-        db.rollback()
-        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
-        db.execute(delete(User).where(User.phone == phone))
-        db.commit()
-        db.close()
-
-
-def test_feed_reward_detail_keeps_each_record_adn_instead_of_parent_adn() -> None:
+def test_feed_reward_details_keep_each_record_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_can_fallback_to_unique_trace_impression_only() -> None:
-    """发奖会话是整场 ID、展示会话是 impressionId 时，只在 trace+eCPM 唯一时回填 ADN。"""
+def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
    db = SessionLocal()
    phone = "18800009995"
+    fallback_date = "2040-02-07"
    try:
        user = User(phone=phone, username="29999999995", register_channel="sms")
        db.add(user)
@@ -368,7 +266,7 @@ def test_feed_reward_source_can_fallback_to_unique_trace_impression_only() -> No
        db.add_all([
            AdFeedRewardRecord(
                client_event_id="source-fallback-unique", user_id=user.id,
-                reward_date=SOURCE_FALLBACK_DATE, duration_seconds=3, unit_count=0,
+                reward_date=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",
@@ -376,7 +274,7 @@ def test_feed_reward_source_can_fallback_to_unique_trace_impression_only() -> No
            ),
            AdFeedRewardRecord(
                client_event_id="source-fallback-ambiguous", user_id=user.id,
-                reward_date=SOURCE_FALLBACK_DATE, duration_seconds=3, unit_count=0,
+                reward_date=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",
@@ -385,31 +283,27 @@ def test_feed_reward_source_can_fallback_to_unique_trace_impression_only() -> No
            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=SOURCE_FALLBACK_DATE,
+                app_env="prod", our_code_id="104098712", report_date=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=SOURCE_FALLBACK_DATE,
+                app_env="prod", our_code_id="104098712", report_date=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=SOURCE_FALLBACK_DATE,
+                app_env="prod", our_code_id="104098712", report_date=fallback_date,
                created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
            ),
        ])
        db.commit()

        result = ad_revenue.ad_revenue_report(
-            db,
-            date_from=SOURCE_FALLBACK_DATE,
-            date_to=SOURCE_FALLBACK_DATE,
-            user_id=user.id,
-            app_env="prod",
-            revenue_scope="all",
+            db, date_from=fallback_date, date_to=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"]}
@@ -419,27 +313,8 @@ def test_feed_reward_source_can_fallback_to_unique_trace_impression_only() -> No
        assert details["4800"]["slot_id"] is None
    finally:
        db.rollback()
-        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(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == fallback_date))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == 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
@@ -0,0 +1,71 @@
+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
@@ -1,7 +1,7 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations

-from datetime import datetime
+from datetime import UTC, datetime

 import pytest
 from fastapi.testclient import TestClient
@@ -9,6 +9,7 @@ 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
@@ -144,6 +145,8 @@ 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:
@@ -162,6 +165,17 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
    assert records.status_code == 200, records.text


+def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
+    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
+    naive = datetime(2038, 1, 1, 8, 0)
+    aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)
+
+    rows = [{"created_at": aware}, {"created_at": naive}]
+    rows.sort(key=queries._coin_record_sort_key, reverse=True)
+
+    assert rows == [{"created_at": naive}, {"created_at": aware}]
+
+
 def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
    _seed_user_with_data("13800000003")
    r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
@@ -9,17 +9,16 @@ 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:
@@ -216,7 +215,9 @@ 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:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
        r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
    assert r.status_code == 200
    with SessionLocal() as db:
@@ -224,6 +225,7 @@ 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:
@@ -0,0 +1,161 @@
+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()
@@ -0,0 +1,17 @@
+"""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
@@ -0,0 +1,102 @@
+"""_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
@@ -132,6 +132,7 @@ 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}},
@@ -139,7 +140,12 @@ 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(compare_record, "fetch_llm_calls", lambda trace_id: sample)
+    monkeypatch.setattr(
+        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
+    )
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )

    db = SessionLocal()
    try:
@@ -0,0 +1,55 @@
+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
Author	SHA1	Message	Date
guke	f0b7f3951e	修复:ComparisonResultIn 补 skipped_dish_count(逐平台缺菜数落库) pricebot 把逐平台缺菜数冗余进 comparison_results 每行,但上报入参 schema ComparisonResultIn 未声明该字段,model_dump() 会在 POST 路径静默丢弃, 记录页三平台网格拿不到逐格「缺少 X 个菜品」。显式声明补齐 + 回归测试。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-27 16:01:50 +08:00
linkeyu	1226bc8365	修复比价记录平均 TOKEN 成本采集与回填 (#182 ) ## 问题原因 - 新版客户端通过服务端 harvest 落库，但完成路径没有触发 LLM 调用明细与 TOKEN 成本回填。 - 内部共享密钥不一致或 PriceBot 实例切换后，拉取失败只留下空值，后续没有自动补偿。 ## 本次改动 - harvest 完成后立即异步回填 LLM 调用、TOKEN 数及成本快照。 - 抽取统一、幂等的成本回填服务，并增加定时补偿 worker。 - 增加 PriceBot 内部鉴权预检、错误日志和多实例查找兜底。 - 仅回填当前价格配置生效后的终态记录，避免用现价误算更早历史数据。 - 补充环境配置、部署说明和单元测试。 ## 验证 - 相关测试：36 passed。 - 静态检查通过，diff check 通过。 - 本地页面显示 ¥0.0139，与数据库精确均值 0.013916 的四舍五入结果一致。 ## 上线注意部署时需确保 app-server 与 PriceBot 的 INTERNAL_API_SECRET 完全一致并重启两个服务；worker 启动后会自动补齐符合条件的历史空值。 --------- Co-authored-by: guke <guke@wonderable.ai> Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #182 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-27 15:51:19 +08:00
linkeyu	22a1105000	修复用户反馈机型与系统版本补全 (#181 ) ## 改动说明 - 反馈列表按同一用户、同一设备编码和反馈提交时间，补全可读机型、厂商及 ROM 大版本 - 补充常见线上机型编码映射 - 更新本地 mock 数据并增加时间边界回归测试 ## 验证 - 反馈相关测试：3 项通过 - Ruff 检查通过 --------- Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #181 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-27 13:33:55 +08:00
guke	b2a528eba1	fix(comparison): 派生 best 排除缺菜店, 避免虚低价当"最低价" (#176 ) pricebot comparison_results[].rank 是纯价格排序(含缺菜店), server 派生 best 若照单全收, 会把缺菜(漏菜)店的虚低总价当 best_price → 记录页戴"最低"红框 + 算出虚假省额。 - _derive / _derive_from_results 派生 best 时按 platform_results[pid]. skipped_dish_count 排除缺菜店; 源平台永远全菜, 全目标缺菜时回落到源 (is_source_best、saved=0), 不虚报省额。 - platform_results 内层结构宽松(老客户端透传可伪造), _is_short 用 isinstance 兜底, 值非 dict 时按"不缺菜"处理, 不打 500。 - harvest_done 传入 done_params.platform_results; 不传→纯 rank/price 老行为不变。 - 新增纯函数测试: 排除缺菜 / 全缺菜回落源 / 不传保持老行为 / 内层非 dict 不崩, 覆盖 _derive 与 _derive_from_results 两条路径。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #176	2026-07-27 10:35:15 +08:00
linkeyu	cdd49c6421	修复：补齐收益明细广告网络来源 (#179 ) ## 本次改动 - 激励视频发奖记录按用户与 `ad_session_id` 回填展示侧 ADN - 信息流明细保留每条发奖自身上报的 ADN 与底层代码位 - 历史记录仅在 `用户 + trace_id + eCPM` 候选网络唯一时安全回填 - 多网络候选保持空值，避免错误归因 ## 验证专项测试 4 passed，覆盖不同 ADN 明细及唯一/歧义回填场景。 --------- Co-authored-by: guke <guke@wonderable.ai> Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #179 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-27 10:24:32 +08:00
linkeyu	775a503d6f	修复广告收益用户详情加载失败 (#180 ) ## 问题原因 - 奖励统计接口用完整 User ORM 判断用户存在，滚动发布或表结构未同步时会因无关字段导致 500 - 金币记录合并广告与签到数据后直接排序，PostgreSQL 中 aware/naive datetime 混排会抛异常 ## 修复内容 - 新增只投影 user.id 的用户存在性检查，保持不存在用户返回 404 - 金币记录排序前统一转换为 aware UTC 排序键 - 增加旧表结构投影与混合时区回归测试 ## 验证结果 - tests/test_admin_read.py：18 项全部通过 - 线上只读数据库回归：近期 7 个活跃用户统计与金币明细全部正常返回；用户 #33 返回 1514 条记录 - 语法/未定义引用检查通过 - 全量测试 532 通过、9 失败；失败均在未修改的 origin/main 基线上复现，和本 PR 无关 --------- Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #180 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-27 10:15:54 +08:00