优化：拆分历史节省金额批量回填

成功但未节省的比价记录纳入平均值分母，无法计算的记录保持排除。新增历史数据迁移，按完整购物篮规则排除漏菜平台并回填最优平台与节省金额。

app/admin/repositories/ad_revenue.py

@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
 _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
 
 
-def business_code_ids(db: Session, app_env: str | None) -> set[str]:
+def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
     """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
     prod_config = app_config.get_ad_config(db)
     prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(
 
     # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
     # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
-    business_ids: set[str] | None = None
+    business_code_ids: set[str] | None = None
     if revenue_scope == "business":
-        business_ids = business_code_ids(db, app_env)
-        events = [e for e in events if e.get("our_code_id") in business_ids]
+        business_code_ids = _business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_code_ids]
 
     # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
     # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
             date_from=date_from,
             date_to=date_to,
             app_env=app_env,
-            our_code_ids=business_ids,
+            our_code_ids=business_code_ids,
         )
         if pangle_aggs:
             by_date = {a["date"]: a for a in pangle_aggs}

app/admin/repositories/queries.py

@@ -14,7 +14,6 @@ from sqlalchemy.orm import Session
 
 from app.core import rewards
 from app.core.config import settings
-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.admin import AdminAuditLog
@@ -1258,15 +1257,11 @@ def user_reward_stats(
     date_from: datetime | None = None,
     date_to: datetime | None = None,
     withdraw_source: str | None = None,
-    app_env: str | None = None,
-    revenue_scope: str = "all",
-    feed_scene: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
-    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
-    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
-    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
+    口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
+    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
     withdraw_source_conds = (
@@ -1301,68 +1296,30 @@ def user_reward_stats(
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
-    business_ids: set[str] | None = None
-    if revenue_scope == "business":
-        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
-        from app.admin.repositories.ad_revenue import business_code_ids
-
-        business_ids = business_code_ids(db, app_env)
-
-    rv_conds = [
-        AdRewardRecord.user_id == user_id,
-        AdRewardRecord.reward_scene == "reward_video",
-        AdRewardRecord.status == "granted",
-        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
-    ]
-    if app_env is not None:
-        rv_conds.append(AdRewardRecord.app_env == app_env)
-    if business_ids is not None:
-        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
     rv = db.execute(
         select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
-            *rv_conds,
+            AdRewardRecord.user_id == user_id,
+            AdRewardRecord.reward_scene == "reward_video",
+            AdRewardRecord.status == "granted",
+            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
         )
     ).all()
     rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
     rv_coins = sum(r.coin for r in rv)
 
-    feed_reward_conds = [
-        AdFeedRewardRecord.user_id == user_id,
-        AdFeedRewardRecord.status == "granted",
-        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
-    ]
-    if app_env is not None:
-        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
-    if feed_scene is not None:
-        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
-    if business_ids is not None:
-        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
-    feed_rewards = db.execute(
+    feed = db.execute(
         select(
             AdFeedRewardRecord.unit_count,
+            AdFeedRewardRecord.ecpm_raw,
             AdFeedRewardRecord.coin,
         ).where(
-            *feed_reward_conds,
+            AdFeedRewardRecord.user_id == user_id,
+            AdFeedRewardRecord.status == "granted",
+            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
         )
     ).all()
-    feed_coins = sum(f.coin for f in feed_rewards)
-
-    feed_impression_conds = [
-        AdEcpmRecord.user_id == user_id,
-        AdEcpmRecord.ad_type.in_(("draw", "feed")),
-        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
-    ]
-    if app_env is not None:
-        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
-    if feed_scene is not None:
-        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
-    if business_ids is not None:
-        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
-    feed_impressions = db.execute(
-        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
-    ).all()
-    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
-    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
+    feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
+    feed_coins = sum(f.coin for f in feed)
 
     trad_coins = db.execute(
         select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1381,7 +1338,7 @@ def user_reward_stats(
         "reward_video_count": len(rv),
         "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
         "reward_video_cash_cents": _coins_to_cents(rv_coins),
-        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
+        "feed_count": int(sum(f.unit_count for f in feed)),
         "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
         "feed_cash_cents": _coins_to_cents(feed_coins),
     }

app/admin/repositories/stats.py

@@ -298,21 +298,31 @@ def dashboard_overview(
         if period_comparison_success_denominator > 0
         else None
     )
-    period_saved_positive_count = _count(
+    # “平均节省金额”表示一次可计算的成功比价平均能省多少钱：
+    # 原平台已经最便宜时 saved_amount_cents=0，也必须进入分母；否则只统计
+    # “确实省到钱”的记录会系统性抬高大盘指标。源价/最优价缺失导致的 NULL
+    # 无法判断实际节省额，仍排除在分母之外。
+    period_saved_calculable_count = _count(
         ComparisonRecord,
         *period_comparison_conds,
         ComparisonRecord.status == "success",
-        ComparisonRecord.saved_amount_cents > 0,
+        ComparisonRecord.saved_amount_cents.is_not(None),
     )
-    period_saved_positive_sum = _sum(
-        ComparisonRecord.saved_amount_cents,
+    period_saved_sum = _sum(
+        case(
+            (
+                ComparisonRecord.saved_amount_cents > 0,
+                ComparisonRecord.saved_amount_cents,
+            ),
+            else_=0,
+        ),
         *period_comparison_conds,
         ComparisonRecord.status == "success",
-        ComparisonRecord.saved_amount_cents > 0,
+        ComparisonRecord.saved_amount_cents.is_not(None),
     )
     period_avg_saved_cents = (
-        round(period_saved_positive_sum / period_saved_positive_count)
-        if period_saved_positive_count
+        round(period_saved_sum / period_saved_calculable_count)
+        if period_saved_calculable_count
         else None
     )
     period_avg_duration_ms = db.execute(

app/admin/routers/users.py

@@ -88,11 +88,6 @@ def get_user_reward_stats(
     withdraw_source: Annotated[
         str | None, Query(pattern="^(coin_cash|invite_cash)$")
     ] = None,
-    app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
-    revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
-    feed_scene: Annotated[
-        str | None, Query(pattern="^(comparison|coupon|welfare)$")
-    ] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
     if not user_repo.user_exists(db, user_id):
@@ -104,9 +99,6 @@ def get_user_reward_stats(
             date_from=date_from,
             date_to=date_to,
             withdraw_source=withdraw_source,
-            app_env=app_env,
-            revenue_scope=revenue_scope,
-            feed_scene=feed_scene,
         )
     )
 

app/admin/schemas/user.py

@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
     reward_video_avg_ecpm: float  # 平均激励视频 eCPM(分/千次)
     reward_video_cash_cents: int  # 激励视频提现(金币折现)
     feed_count: int  # 累计信息流广告数(granted 份数,unit_count 累加)
-    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
+    feed_avg_ecpm: float  # 平均信息流广告 eCPM(分/千次)
     feed_cash_cents: int  # 信息流广告提现(金币折现)
 
 

app/api/v1/compare.py

@@ -108,7 +108,7 @@ def _harvest_done_blocking(
 
 def _harvest_abort_blocking(
     trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
-) -> int | None:
+) -> None:
     with SessionLocal() as db:
         rec = crud_compare.harvest_abort(
             db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
@@ -118,7 +118,6 @@ def _harvest_abort_blocking(
         extra={"phase": "harvest_abort",
                "status": (rec.status if rec else None), "reason": reason},
     )
-    return rec.id if rec is not None else None
 
 
 async def _forward(
@@ -292,10 +291,7 @@ async def trace_epilogue(
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
 async def trace_finalize(
-    request: Request,
-    background_tasks: BackgroundTasks,
-    user: OptionalUser,
-    db: DbSession,
+    request: Request, user: OptionalUser, db: DbSession
 ) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -306,16 +302,12 @@ async def trace_finalize(
         request, "/api/trace/finalize", user, harvest_first_frame=False,
     )
     try:
-        record_id = await run_in_threadpool(
+        await run_in_threadpool(
             _harvest_abort_blocking, trace_id,
             (meta.get("status") or "cancelled"),
             (meta.get("reason") or meta.get("information")),
             (resp.get("trace_url") if isinstance(resp, dict) else None),
         )
-        if record_id is not None:
-            background_tasks.add_task(
-                backfill_comparison_llm_cost, record_id, trace_id
-            )
     except Exception as e:  # noqa: BLE001
         logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
     return resp

app/services/comparison_llm_backfill.py

@@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs(
                 select(ComparisonRecord.id, ComparisonRecord.trace_id)
                 .where(
                     *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed", "cancelled")),
+                    ComparisonRecord.status.in_(("success", "failed")),
                     ComparisonRecord.llm_cost_yuan.is_(None),
                 )
                 .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())

app/services/comparison_savings_backfill.py

@@ -0,0 +1,201 @@
+"""分批修正历史比价记录的最优平台与节省金额派生列。
+
+该回填刻意不放进 Alembic：生产启动迁移不应一次性读取并更新全部历史记录。
+调用方按主键游标分批执行，每批独立事务；默认 dry-run 时只计算差异、不写库。
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models.comparison import ComparisonRecord
+
+_DERIVED_FIELDS = (
+    "source_price_cents",
+    "best_platform_id",
+    "best_platform_name",
+    "best_price_cents",
+    "saved_amount_cents",
+    "is_source_best",
+)
+
+
+@dataclass(frozen=True)
+class SavingsBackfillBatch:
+    """一批回填结果；last_id 可直接作为下一批 after_id。"""
+
+    scanned: int
+    changed: int
+    last_id: int
+    done: bool
+
+
+def _json_value(value: Any, fallback: Any) -> Any:
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except (TypeError, ValueError):
+            return fallback
+    return value
+
+
+def _decimal(value: Any) -> Decimal | None:
+    if value is None or isinstance(value, bool):
+        return None
+    try:
+        return Decimal(str(value))
+    except (InvalidOperation, TypeError, ValueError):
+        return None
+
+
+def _yuan_to_cents(value: Any) -> int | None:
+    amount = _decimal(value)
+    if amount is None:
+        return None
+    return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
+
+
+def _rank(value: Any) -> Decimal:
+    rank = _decimal(value)
+    return rank if rank is not None else Decimal("1e18")
+
+
+def derive_historical_savings(
+    source_price_cents: int | None,
+    comparison_results: Any,
+    raw_payload: Any,
+) -> dict[str, Any] | None:
+    """按当前完整购物篮规则，从历史原始数据重新派生结构化金额字段。"""
+    results = _json_value(comparison_results, [])
+    raw = _json_value(raw_payload, {})
+    if not isinstance(results, list):
+        return None
+    platform_results = raw.get("platform_results", {}) if isinstance(raw, dict) else {}
+    if not isinstance(platform_results, dict):
+        platform_results = {}
+
+    clean: list[tuple[dict[str, Any], Decimal]] = []
+    for result in results:
+        if not isinstance(result, dict):
+            continue
+        price = _decimal(result.get("price"))
+        if price is None:
+            continue
+        platform_id = result.get("platform_id")
+        platform_info = platform_results.get(platform_id) if platform_id else None
+        skipped_count = (
+            _decimal(platform_info.get("skipped_dish_count"))
+            if isinstance(platform_info, dict)
+            else None
+        )
+        if skipped_count is not None and skipped_count > 0:
+            continue
+        clean.append((result, price))
+
+    if not clean:
+        return None
+
+    best, best_price = min(
+        clean,
+        key=lambda item: (_rank(item[0].get("rank")), item[1]),
+    )
+    if source_price_cents is None:
+        source = next(
+            (
+                result
+                for result in results
+                if isinstance(result, dict)
+                and result.get("is_source")
+                and _decimal(result.get("price")) is not None
+            ),
+            None,
+        )
+        source_price_cents = _yuan_to_cents(source.get("price")) if source else None
+
+    best_price_cents = _yuan_to_cents(best_price)
+    saved_amount_cents = (
+        source_price_cents - best_price_cents
+        if source_price_cents is not None and best_price_cents is not None
+        else None
+    )
+    return {
+        "source_price_cents": source_price_cents,
+        "best_platform_id": best.get("platform_id"),
+        "best_platform_name": best.get("platform_name"),
+        "best_price_cents": best_price_cents,
+        "saved_amount_cents": saved_amount_cents,
+        "is_source_best": bool(best.get("is_source")),
+    }
+
+
+def latest_success_id(db: Session) -> int:
+    """固定本次任务的扫描上界，避免执行期间新增记录让任务无限延长。"""
+    return int(
+        db.execute(
+            select(func.coalesce(func.max(ComparisonRecord.id), 0)).where(
+                ComparisonRecord.status == "success"
+            )
+        ).scalar_one()
+    )
+
+
+def repair_comparison_savings_batch(
+    db: Session,
+    *,
+    after_id: int,
+    max_id: int,
+    batch_size: int = 500,
+    apply: bool = False,
+) -> SavingsBackfillBatch:
+    """按 id 游标处理一批；apply=True 时仅更新实际发生变化的记录并提交。"""
+    if after_id < 0 or max_id < 0:
+        raise ValueError("after_id and max_id must be non-negative")
+    if batch_size <= 0:
+        raise ValueError("batch_size must be positive")
+
+    records = list(
+        db.scalars(
+            select(ComparisonRecord)
+            .where(
+                ComparisonRecord.status == "success",
+                ComparisonRecord.id > after_id,
+                ComparisonRecord.id <= max_id,
+            )
+            .order_by(ComparisonRecord.id)
+            .limit(batch_size)
+        )
+    )
+    changed = 0
+    for record in records:
+        derived = derive_historical_savings(
+            record.source_price_cents,
+            record.comparison_results,
+            record.raw_payload,
+        )
+        if derived is None:
+            continue
+        differs = any(
+            getattr(record, field) != derived[field] for field in _DERIVED_FIELDS
+        )
+        if not differs:
+            continue
+        changed += 1
+        if apply:
+            for field in _DERIVED_FIELDS:
+                setattr(record, field, derived[field])
+
+    if apply:
+        db.commit()
+
+    last_id = records[-1].id if records else after_id
+    return SavingsBackfillBatch(
+        scanned=len(records),
+        changed=changed,
+        last_id=last_id,
+        done=not records or len(records) < batch_size or last_id >= max_id,
+    )

scripts/backfill_comparison_savings.py

@@ -0,0 +1,103 @@
+"""分批回填历史比价节省金额。
+
+默认只预览，不写库：
+    python -m scripts.backfill_comparison_savings
+
+确认预览后实际执行：
+    python -m scripts.backfill_comparison_savings --apply
+
+中断后可从日志最后一个 last_id 继续：
+    python -m scripts.backfill_comparison_savings --apply --after-id 12345
+"""
+from __future__ import annotations
+
+import argparse
+
+from app.db.session import SessionLocal
+from app.services.comparison_savings_backfill import (
+    latest_success_id,
+    repair_comparison_savings_batch,
+)
+
+
+def _positive_int(value: str) -> int:
+    parsed = int(value)
+    if parsed <= 0:
+        raise argparse.ArgumentTypeError("必须为正整数")
+    return parsed
+
+
+def _non_negative_int(value: str) -> int:
+    parsed = int(value)
+    if parsed < 0:
+        raise argparse.ArgumentTypeError("不能为负数")
+    return parsed
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="按完整购物篮口径分批回填历史比价节省金额（默认 dry-run）"
+    )
+    parser.add_argument("--apply", action="store_true", help="实际写库；默认仅预览")
+    parser.add_argument("--batch-size", type=_positive_int, default=500)
+    parser.add_argument(
+        "--after-id",
+        type=_non_negative_int,
+        default=0,
+        help="只处理大于该主键的记录，用于断点续跑",
+    )
+    parser.add_argument(
+        "--max-id",
+        type=_non_negative_int,
+        help="固定扫描上界；不传则启动时取成功记录最大主键",
+    )
+    parser.add_argument(
+        "--max-batches",
+        type=_positive_int,
+        help="本次最多处理多少批，便于灰度限量执行",
+    )
+    args = parser.parse_args()
+
+    with SessionLocal() as db:
+        max_id = args.max_id if args.max_id is not None else latest_success_id(db)
+
+    mode = "APPLY" if args.apply else "DRY-RUN"
+    after_id = args.after_id
+    scanned = 0
+    changed = 0
+    batches = 0
+    print(
+        f"comparison savings backfill mode={mode} "
+        f"after_id={after_id} max_id={max_id} batch_size={args.batch_size}"
+    )
+
+    while after_id < max_id:
+        with SessionLocal() as db:
+            result = repair_comparison_savings_batch(
+                db,
+                after_id=after_id,
+                max_id=max_id,
+                batch_size=args.batch_size,
+                apply=args.apply,
+            )
+        batches += 1
+        scanned += result.scanned
+        changed += result.changed
+        after_id = result.last_id
+        print(
+            f"batch={batches} scanned={result.scanned} changed={result.changed} "
+            f"last_id={result.last_id} done={result.done}"
+        )
+        if result.done or (
+            args.max_batches is not None and batches >= args.max_batches
+        ):
+            break
+
+    print(
+        f"completed mode={mode} batches={batches} scanned={scanned} "
+        f"changed={changed} last_id={after_id} max_id={max_id}"
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/test_admin_read.py

@@ -120,6 +120,55 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
     assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
 
 
+def test_dashboard_average_saved_includes_zero_and_excludes_uncalculable(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    created_at = datetime(2037, 2, 15, 12)
+    db = SessionLocal()
+    try:
+        db.add_all(
+            [
+                ComparisonRecord(
+                    trace_id="dashboard-saved-positive",
+                    status="success",
+                    saved_amount_cents=100,
+                    created_at=created_at,
+                ),
+                ComparisonRecord(
+                    trace_id="dashboard-saved-source-best",
+                    status="success",
+                    saved_amount_cents=0,
+                    created_at=created_at,
+                ),
+                ComparisonRecord(
+                    trace_id="dashboard-saved-uncalculable",
+                    status="success",
+                    saved_amount_cents=None,
+                    created_at=created_at,
+                ),
+                ComparisonRecord(
+                    trace_id="dashboard-saved-failed",
+                    status="failed",
+                    saved_amount_cents=900,
+                    created_at=created_at,
+                ),
+            ]
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2037-02-15", "date_to": "2037-02-15"},
+        headers=_auth(admin_token),
+    )
+
+    assert response.status_code == 200, response.text
+    # (100 + 0) / 2；NULL 无法计算，failed 不属于成功比价。
+    assert response.json()["period"]["comparison"]["average_saved_cents"] == 50
+
+
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
     uid = _seed_user_with_data("13800000002")
     r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
@@ -219,115 +268,6 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
     assert invite.json()["cash_balance_cents"] == 456
 
 
-def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
-    from app.models.ad_ecpm import AdEcpmRecord
-    from app.models.ad_feed_reward import AdFeedRewardRecord
-
-    uid = _seed_user_with_data("13800000024")
-    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
-    db = SessionLocal()
-    try:
-        db.add_all(
-            [
-                AdFeedRewardRecord(
-                    client_event_id="reward-stats-granted-high",
-                    ad_session_id="reward-stats-granted-high",
-                    user_id=uid,
-                    reward_date="2038-01-15",
-                    duration_seconds=10,
-                    unit_count=1,
-                    ecpm_raw="9000",
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    coin=9,
-                    status="granted",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-impression-low",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    ecpm_raw="1000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="feed",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-impression-mid",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    ecpm_raw="3000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="comparison",
-                    ad_session_id="reward-stats-other-scene",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    ecpm_raw="7000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-test-env",
-                    app_env="test",
-                    our_code_id="104127529",
-                    ecpm_raw="8000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-non-business",
-                    app_env="prod",
-                    our_code_id="demo-slot",
-                    ecpm_raw="9000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-            ]
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    response = admin_client.get(
-        f"/admin/api/users/{uid}/reward-stats",
-        params={
-            "date_from": "2038-01-15T00:00:00Z",
-            "date_to": "2038-01-15T23:59:59Z",
-            "app_env": "prod",
-            "revenue_scope": "business",
-            "feed_scene": "coupon",
-        },
-        headers=_auth(admin_token),
-    )
-    assert response.status_code == 200, response.text
-    data = response.json()
-    assert data["feed_count"] == 1
-    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
-    assert data["feed_avg_ecpm"] == 2000.0
-
-
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
     """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
     naive = datetime(2038, 1, 1, 8, 0)

tests/test_compare_harvest.py

@@ -263,9 +263,7 @@ def test_trace_finalize_harvests_abort(client) -> None:
     with SessionLocal() as db:                        # 先有 running 行(帧0建的)
         crud.harvest_running(db, trace_id=tid, user_id=None)
     p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"})
-    with p, patch(
-        "app.api.v1.compare.backfill_comparison_llm_cost"
-    ) as backfill:
+    with p:
         r = client.post("/api/v1/trace/finalize",
                         json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
     assert r.status_code == 200
@@ -273,7 +271,6 @@ def test_trace_finalize_harvests_abort(client) -> None:
         rec = _get(db, tid)
         assert rec is not None and rec.status == "cancelled"
         assert rec.trace_url.endswith("/fin/")
-        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_price_step_binds_user_when_authed(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -72,7 +72,6 @@ def test_backfill_retries_then_persists_cost(monkeypatch):
 
 def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
     missing_id = _record("llm-repair-missing")
-    cancelled_id = _record("llm-repair-cancelled", status="cancelled")
     running_id = _record("llm-repair-running", status="running")
     calls = [
         {
@@ -97,15 +96,12 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
         )
         assert result["repaired"] >= 1
         assert "llm-repair-missing" in seen
-        assert "llm-repair-cancelled" 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, cancelled_id).llm_cost_yuan is not None
             assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
     finally:
         _delete(missing_id)
-        _delete(cancelled_id)
         _delete(running_id)
 
 

tests/test_comparison_savings_backfill.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
+from app.services.comparison_savings_backfill import (
+    latest_success_id,
+    repair_comparison_savings_batch,
+)
+
+
+def _record(trace_id: str, *, status: str = "success") -> ComparisonRecord:
+    return ComparisonRecord(
+        trace_id=trace_id,
+        status=status,
+        source_price_cents=4200,
+        best_platform_id="missing-dishes",
+        best_platform_name="Incomplete",
+        best_price_cents=2500,
+        saved_amount_cents=1700,
+        is_source_best=False,
+        comparison_results=[
+            {
+                "platform_id": "source",
+                "platform_name": "Source",
+                "price": 42,
+                "rank": 3,
+                "is_source": True,
+            },
+            {
+                "platform_id": "missing-dishes",
+                "platform_name": "Incomplete",
+                "price": 25,
+                "rank": 1,
+                "is_source": False,
+            },
+            {
+                "platform_id": "complete",
+                "platform_name": "Complete",
+                "price": 38.5,
+                "rank": 2,
+                "is_source": False,
+            },
+        ],
+        raw_payload={
+            "platform_results": {
+                "missing-dishes": {"skipped_dish_count": 1},
+                "complete": {"skipped_dish_count": 0},
+            }
+        },
+        created_at=datetime(2037, 3, 1, 12),
+    )
+
+
+def test_comparison_savings_backfill_is_batched_dry_run_and_idempotent() -> None:
+    db = SessionLocal()
+    try:
+        affected = _record("savings-backfill-affected")
+        unchanged = _record("savings-backfill-unchanged")
+        unchanged.best_platform_id = "complete"
+        unchanged.best_platform_name = "Complete"
+        unchanged.best_price_cents = 3850
+        unchanged.saved_amount_cents = 350
+        failed = _record("savings-backfill-failed", status="failed")
+        db.add_all([affected, unchanged, failed])
+        db.commit()
+        affected_id = affected.id
+        unchanged_id = unchanged.id
+        failed_id = failed.id
+    finally:
+        db.close()
+
+    db = SessionLocal()
+    try:
+        assert latest_success_id(db) >= unchanged_id
+        max_id = unchanged_id
+        affected = db.get(ComparisonRecord, affected_id)
+        assert affected is not None
+
+        preview = repair_comparison_savings_batch(
+            db,
+            after_id=affected_id - 1,
+            max_id=max_id,
+            batch_size=1,
+        )
+        assert preview.scanned == 1
+        assert preview.changed == 1
+        assert preview.last_id == affected_id
+        assert not preview.done
+        db.refresh(affected)
+        assert affected.best_platform_id == "missing-dishes"
+        assert affected.saved_amount_cents == 1700
+
+        applied = repair_comparison_savings_batch(
+            db,
+            after_id=affected_id - 1,
+            max_id=max_id,
+            batch_size=1,
+            apply=True,
+        )
+        assert applied.changed == 1
+        db.refresh(affected)
+        assert affected.best_platform_id == "complete"
+        assert affected.best_platform_name == "Complete"
+        assert affected.best_price_cents == 3850
+        assert affected.saved_amount_cents == 350
+
+        second_batch = repair_comparison_savings_batch(
+            db,
+            after_id=applied.last_id,
+            max_id=max_id,
+            batch_size=1,
+            apply=True,
+        )
+        assert second_batch.scanned == 1
+        assert second_batch.changed == 0
+        assert second_batch.last_id == unchanged_id
+        assert second_batch.done
+
+        rerun = repair_comparison_savings_batch(
+            db,
+            after_id=affected_id - 1,
+            max_id=max_id,
+            batch_size=10,
+            apply=True,
+        )
+        assert rerun.scanned == 2
+        assert rerun.changed == 0
+
+        failed_row = db.get(ComparisonRecord, failed_id)
+        assert failed_row is not None
+        assert failed_row.best_platform_id == "missing-dishes"
+        assert failed_row.saved_amount_cents == 1700
+    finally:
+        db.close()