修复：补齐用户领券记录点位统计

Reviewed-on: #195

alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py

@@ -0,0 +1,26 @@
+"""merge comparison platforms + fail_reason heads
+
+Revision ID: 6d2309208549
+Revises: comparison_platforms_col, comparison_record_fail_reason
+Create Date: 2026-07-29 01:48:41.868083
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '6d2309208549'
+down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

alembic/versions/comparison_platforms_col.py

@@ -0,0 +1,47 @@
+"""add platforms unified array column to comparison_record
+
+展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
+status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
+纯新增列, 老记录为空 → 前端回退老 comparison_results。
+
+Revision ID: comparison_platforms_col
+Revises: user_manual_risk_fields
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "comparison_platforms_col"
+down_revision: str | None = "user_manual_risk_fields"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+
+def upgrade() -> None:
+    # 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
+    # `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
+    #  NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
+    # 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
+    bind = op.get_bind()
+    cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
+    if "platforms" in cols:
+        return
+    with op.batch_alter_table("comparison_record") as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "platforms", _JSON, nullable=False,
+                server_default=sa.text("'[]'"),
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("comparison_record") as batch_op:
+        batch_op.drop_column("platforms")

alembic/versions/coupon_session_planned_count.py

@@ -0,0 +1,28 @@
+"""add planned coupon count to coupon session
+
+Revision ID: coupon_session_planned_count
+Revises: 6d2309208549
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "coupon_session_planned_count"
+down_revision: str | None = "6d2309208549"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "coupon_session",
+        sa.Column("planned_coupon_count", sa.Integer(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("coupon_session", "planned_coupon_count")

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_code_ids: set[str] | None = None
+    business_ids: set[str] | None = None
     if revenue_scope == "business":
-        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]
+        business_ids = business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_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_code_ids,
+            our_code_ids=business_ids,
         )
         if pangle_aggs:
             by_date = {a["date"]: a for a in pangle_aggs}

app/admin/repositories/coupon_data.py

@@ -23,6 +23,7 @@ from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platfo
 
 _SLOT_OK = ("success", "already_claimed")
 _SLOT_TRIED = ("success", "already_claimed", "failed")
+_SESSION_POINT_TOTAL = (*_SLOT_TRIED, "skipped")
 
 
 def _cn_hour(dt: datetime) -> int:
@@ -166,6 +167,10 @@ def _session_to_row(
     point_stats: dict | None = None,
 ) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
+    succeeded = point_stats["succeeded"] if point_stats else 0
+    event_total = point_stats["tried"] if point_stats else 0
+    planned_total = r.planned_coupon_count or 0
+    point_total = max(event_total, planned_total)
     return {
         "id": r.id,
         "trace_id": r.trace_id,
@@ -182,8 +187,8 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": point_stats["succeeded"] if point_stats else None,
-        "point_total_count": point_stats["tried"] if point_stats else None,
+        "point_success_count": succeeded if point_total > 0 else None,
+        "point_total_count": point_total if point_total > 0 else None,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
@@ -202,7 +207,7 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
         )
         .where(
             CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimEvent.status.in_(_SLOT_TRIED),
+            CouponClaimEvent.status.in_(_SESSION_POINT_TOTAL),
         )
         .group_by(CouponClaimEvent.trace_id)
     ).all()
@@ -401,11 +406,13 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
         select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
     ).scalar_one()
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
+    point_stats_map = _point_scores_by_trace(db, [r.trace_id for r in rows])
     return {
         "items": [
             _session_to_row(
                 r,
                 ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
+                point_stats=point_stats_map.get(r.trace_id),
             )
             for r in rows
         ],

app/admin/repositories/queries.py

@@ -14,6 +14,7 @@ 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
@@ -1257,11 +1258,15 @@ 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 累加);
-    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
+    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
+    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
+    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
     withdraw_source_conds = (
@@ -1296,30 +1301,68 @@ 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(
-            AdRewardRecord.user_id == user_id,
-            AdRewardRecord.reward_scene == "reward_video",
-            AdRewardRecord.status == "granted",
-            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+            *rv_conds,
         )
     ).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 = db.execute(
+    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(
         select(
             AdFeedRewardRecord.unit_count,
-            AdFeedRewardRecord.ecpm_raw,
             AdFeedRewardRecord.coin,
         ).where(
-            AdFeedRewardRecord.user_id == user_id,
-            AdFeedRewardRecord.status == "granted",
-            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+            *feed_reward_conds,
         )
     ).all()
-    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)
+    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]
 
     trad_coins = db.execute(
         select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1338,7 +1381,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)),
+        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
         "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,31 +298,21 @@ def dashboard_overview(
         if period_comparison_success_denominator > 0
         else None
     )
-    # “平均节省金额”表示一次可计算的成功比价平均能省多少钱：
-    # 原平台已经最便宜时 saved_amount_cents=0，也必须进入分母；否则只统计
-    # “确实省到钱”的记录会系统性抬高大盘指标。源价/最优价缺失导致的 NULL
-    # 无法判断实际节省额，仍排除在分母之外。
-    period_saved_calculable_count = _count(
+    period_saved_positive_count = _count(
         ComparisonRecord,
         *period_comparison_conds,
         ComparisonRecord.status == "success",
-        ComparisonRecord.saved_amount_cents.is_not(None),
+        ComparisonRecord.saved_amount_cents > 0,
     )
-    period_saved_sum = _sum(
-        case(
-            (
-                ComparisonRecord.saved_amount_cents > 0,
-                ComparisonRecord.saved_amount_cents,
-            ),
-            else_=0,
-        ),
+    period_saved_positive_sum = _sum(
+        ComparisonRecord.saved_amount_cents,
         *period_comparison_conds,
         ComparisonRecord.status == "success",
-        ComparisonRecord.saved_amount_cents.is_not(None),
+        ComparisonRecord.saved_amount_cents > 0,
     )
     period_avg_saved_cents = (
-        round(period_saved_sum / period_saved_calculable_count)
-        if period_saved_calculable_count
+        round(period_saved_positive_sum / period_saved_positive_count)
+        if period_saved_positive_count
         else None
     )
     period_avg_duration_ms = db.execute(

app/admin/routers/users.py

@@ -88,6 +88,11 @@ 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):
@@ -99,6 +104,9 @@ 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/coupon_data.py

@@ -79,10 +79,15 @@ class CouponDataRow(BaseModel):
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
     point_success_count: int | None = Field(
-        None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
+        None,
+        description="本次成功单券数(success+already_claimed);有计划数但尚无成功事件时为 0",
     )
     point_total_count: int | None = Field(
-        None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
+        None,
+        description=(
+            "本次计划单券数(max(逐券事件数,计划数),事件含 skipped);"
+            "无计划数且无逐券事件时为空"
+        ),
     )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(

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  # 平均信息流广告 eCPM(分/千次)
+    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 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,
-) -> None:
+) -> int | 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,6 +118,7 @@ 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(
@@ -291,7 +292,10 @@ async def trace_epilogue(
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
 async def trace_finalize(
-    request: Request, user: OptionalUser, db: DbSession
+    request: Request,
+    background_tasks: BackgroundTasks,
+    user: OptionalUser,
+    db: DbSession,
 ) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -302,12 +306,16 @@ async def trace_finalize(
         request, "/api/trace/finalize", user, harvest_first_frame=False,
     )
     try:
-        await run_in_threadpool(
+        record_id = 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/api/v1/coupon.py

@@ -93,6 +93,13 @@ def _record_claims_blocking(
             )
 
 
+def _merge_planned_count_blocking(
+    trace_id: str | None, planned_count: int | None
+) -> None:
+    with SessionLocal() as db:
+        coupon_repo.merge_session_planned_count(db, trace_id, planned_count)
+
+
 def _mark_completed_blocking(
     device_id: str, user_id: int | None, trace_id: str | None
 ) -> None:
@@ -176,6 +183,18 @@ async def coupon_step(
 
     resp_json = resp.json()
 
+    # pricebot 每帧 status.progress.total 都带本轮计划券数。独立回写 session 后，
+    # 即使用户在第一张出结果前退出，admin 也能显示 0/N，而不是空值。
+    progress = (resp_json.get("status") or {}).get("progress") or {}
+    planned_count = progress.get("total")
+    if trace_id and isinstance(planned_count, int) and planned_count > 0:
+        try:
+            await run_in_threadpool(
+                _merge_planned_count_blocking, trace_id, planned_count
+            )
+        except Exception as e:  # noqa: BLE001
+            logger.warning("coupon planned count write failed: %s", e)
+
     # 领券结果沉淀:每日资产 + 逐次事件；中间帧和 done 全量帧均幂等写库。
     # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
     if device_id:

app/models/comparison.py

@@ -113,6 +113,10 @@ class ComparisonRecord(Base):
     items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
     comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
+    # 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
+    # status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
+    # + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
+    platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 目标平台未找到、跳过的菜名
     skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底

app/models/coupon_state.py

@@ -277,6 +277,9 @@ class CouponSession(Base):
     platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
     # 领到总张数(收尾帧带)。
     claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # pricebot 本轮计划处理的券数。由 /coupon/step 每帧 status.progress.total 回写；
+    # 即使用户中途退出、尚无任何单券终态，admin 也能以计划数作为成功率分母。
+    planned_coupon_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     # 本次 session 至少领到一张(status∈{success,already_claimed})的平台 id 列表,如 ["meituan-waimai","jd-waimai"]。
     # admin「领券数据」据此算整单成功率(②)/点位成功率(③);服务端 /step 逐帧按 trace_id 并集写入
     # (见 coupon_state.merge_session_platform_success)。旧行=NULL → 视作空集。

app/repositories/comparison.py

@@ -164,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict:
 
     is_source_best = best.is_source if best is not None else None
 
-    # status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
-    status = payload.status
+    # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
+    # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
+    status = payload.record_status or payload.status
     if status is None:
         has_valid_target = any(
             (not r.is_source) and r.price is not None for r in results
@@ -197,16 +198,35 @@ def upsert_record(
     灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
     新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
     """
-    derived = _derive(payload)
+    # 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
+    # _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
+    if payload.platforms:
+        derived = _derive_from_platforms(payload.platforms, payload.record_status)
+        # 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
+        derived["fail_reason"] = (
+            _derive_fail_display(payload.information, payload.platform_results or {})
+            if derived["status"] == "failed"
+            else None
+        )
+        # 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
+        # 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
+        # 与 _derive_from_platforms 同名键撞键 TypeError)。
+        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
+            if not derived.get(_k):
+                derived[_k] = getattr(payload, _k)
+    else:
+        derived = _derive(payload)
+        # _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
+        # 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
+        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
+            derived[_k] = getattr(payload, _k)
     items = [it.model_dump(exclude_none=True) for it in payload.items]
     fields = dict(
         device_id=payload.device_id,
         business_type=payload.business_type,
-        store_name=payload.store_name,
         product_names=_product_names_from_items(items),
-        source_platform_id=payload.source_platform_id,
-        source_platform_name=payload.source_platform_name,
-        source_package=payload.source_package,
+        # store_name / source_platform_id / source_platform_name / source_package 统一由
+        # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
         information=payload.information,
         best_deeplink=payload.best_deeplink,
         trace_url=payload.trace_url,
@@ -214,6 +234,7 @@ def upsert_record(
         skipped_dish_count=payload.skipped_dish_count,
         items=items,
         comparison_results=[r.model_dump() for r in payload.comparison_results],
+        platforms=list(payload.platforms or []),
         skipped_dish_names=list(payload.skipped_dish_names),
         # 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
         device_model=payload.device_model,
@@ -286,7 +307,8 @@ def upsert_record(
 
 
 def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None
+    results: list[dict], platform_results: dict | None = None,
+    record_status: str | None = None,
 ) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
     等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -334,7 +356,53 @@ def _derive_from_results(
         "saved_amount_cents": saved_amount_cents,
         "is_source_best": best.get("is_source") if best else None,
         "store_name": (src_row or {}).get("store_name") or None,
-        "status": "success" if has_valid_target else "failed",
+        # 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
+        # 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
+        # 回退老的 success/failed 二态派生, 向后兼容。
+        "status": record_status or ("success" if has_valid_target else "failed"),
+    }
+
+
+def _derive_from_platforms(
+    platforms: list, record_status: str | None = None,
+) -> dict:
+    """从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
+
+    best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
+    platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
+    消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
+    platforms 时调用方回退 _derive_from_results(向后兼容)。"""
+    rows = [p for p in (platforms or []) if isinstance(p, dict)]
+    src = next((p for p in rows if p.get("role") == "source"), None)
+    best = next((p for p in rows if p.get("is_best")), None)
+    source_price_cents = _yuan_to_cents(src.get("price")) if src else None
+    best_price_cents = _yuan_to_cents(best.get("price")) if best else None
+    saved_amount_cents = None
+    if source_price_cents is not None and best_price_cents is not None:
+        saved_amount_cents = source_price_cents - best_price_cents
+    has_valid_target = any(
+        p.get("role") != "source" and p.get("price") is not None for p in rows
+    )
+    # store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
+    # _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
+    # 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
+    store_name = (
+        (src or {}).get("store_name")
+        or (best or {}).get("store_name")
+        or next((p.get("store_name") for p in rows if p.get("store_name")), None)
+    )
+    return {
+        "source_platform_id": (src or {}).get("platform_id"),
+        "source_platform_name": (src or {}).get("platform_name"),
+        "source_package": (src or {}).get("package"),
+        "source_price_cents": source_price_cents,
+        "best_platform_id": (best or {}).get("platform_id"),
+        "best_platform_name": (best or {}).get("platform_name"),
+        "best_price_cents": best_price_cents,
+        "saved_amount_cents": saved_amount_cents,
+        "is_source_best": (best.get("role") == "source") if best else None,
+        "store_name": store_name or None,
+        "status": record_status or ("success" if has_valid_target else "failed"),
     }
 
 
@@ -493,7 +561,29 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
+    # 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
+    # record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
+    platforms = done_params.get("platforms") or []
+    record_status = done_params.get("record_status")
+    # 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
+    # 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
+    # platforms 时回退从 comparison_results 派生(向后兼容)。
+    if platforms:
+        derived = _derive_from_platforms(platforms, record_status)
+        # 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
+        _item_row = (
+            next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
+            or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
+            or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
+        )
+        items = (_item_row or {}).get("items") or []
+    else:
+        derived = _derive_from_results(
+            results, done_params.get("platform_results"), record_status
+        )
+        # pricebot 已把源单菜品塞进 comparison_results[源行].items
+        items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
+    # 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
     fail_reason = (
         _derive_fail_display(
             done_params.get("information"), done_params.get("platform_results")
@@ -501,8 +591,6 @@ def harvest_done(
         if derived["status"] == "failed"
         else None
     )
-    # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
-    items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(
         business_type=business_type or "food",
         information=done_params.get("information") or None,
@@ -514,6 +602,7 @@ def harvest_done(
         skipped_dish_count=done_params.get("skipped_dish_count"),
         skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
         comparison_results=results,
+        platforms=platforms,
         items=items,
         product_names=_product_names_from_items(items),
         raw_payload=done_params,

app/repositories/coupon_state.py

@@ -157,6 +157,22 @@ def session_app_env(db: Session, trace_id: str | None) -> str | None:
     ).scalar_one_or_none()
 
 
+def merge_session_planned_count(
+    db: Session, trace_id: str | None, planned_count: int | None
+) -> None:
+    """把 pricebot 队列总数回写 session，供中途退出/全跳过场次计算逐券分母。"""
+    if not trace_id or planned_count is None or planned_count <= 0:
+        return
+    row = db.execute(
+        select(CouponSession).where(CouponSession.trace_id == trace_id)
+    ).scalar_one_or_none()
+    if row is None:
+        return
+    # 队列在一个 trace 内固定；取大值可防乱序旧帧覆盖，也兼容调度层补入预跳券。
+    row.planned_coupon_count = max(row.planned_coupon_count or 0, planned_count)
+    db.commit()
+
+
 def record_claims(
     db: Session,
     device_id: str,

app/schemas/compare_record.py

@@ -107,6 +107,13 @@ class ComparisonRecordIn(BaseModel):
     # 明细
     items: list[ComparisonItemIn] = Field(default_factory=list)
     comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
+    # 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
+    # status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
+    # pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
+    platforms: list[dict] = Field(default_factory=list)
+    # 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
+    # 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
+    record_status: str | None = None
     # 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
     # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
     # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -177,6 +184,9 @@ class ComparisonRecordOut(BaseModel):
     fail_reason: str | None = None
     items: list = []
     comparison_results: list = []
+    # 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
+    # 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
+    platforms: list = []
     skipped_dish_names: list = []
     total_ms: int | None = None
     # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。

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")),
+                    ComparisonRecord.status.in_(("success", "failed", "cancelled")),
                     ComparisonRecord.llm_cost_yuan.is_(None),
                 )
                 .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())

app/services/comparison_savings_backfill.py

@@ -1,201 +0,0 @@
-"""分批修正历史比价记录的最优平台与节省金额派生列。
-
-该回填刻意不放进 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

@@ -1,103 +0,0 @@
-"""分批回填历史比价节省金额。
-
-默认只预览，不写库：
-    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,55 +120,6 @@ 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))
@@ -268,6 +219,115 @@ 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,7 +263,9 @@ 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:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
         r = client.post("/api/v1/trace/finalize",
                         json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
     assert r.status_code == 200
@@ -271,6 +273,7 @@ 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,6 +72,7 @@ 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 = [
         {
@@ -96,12 +97,15 @@ 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

@@ -1,136 +0,0 @@
-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()

tests/test_coupon_point_score.py

@@ -10,14 +10,16 @@ from app.admin.repositories.coupon_data import (
     _point_scores_by_trace,
     coupon_data_report,
     coupon_point_details,
+    coupon_user_records,
 )
 from app.admin.security import create_admin_token
 from app.db.session import SessionLocal
 from app.models.coupon_state import CouponClaimEvent, CouponSession
+from app.repositories.coupon_state import merge_session_planned_count
 
 
 def test_point_scores_by_trace() -> None:
-    """已领算成功、失败算尝试、跳过不进分母。"""
+    """已领算成功；失败和跳过均属于本轮计划券，进入分母。"""
     db = SessionLocal()
     trace = "point-score-trace"
     try:
@@ -37,7 +39,7 @@ def test_point_scores_by_trace() -> None:
 
         stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
-        assert stats["tried"] == 3
+        assert stats["tried"] == 4
         details = coupon_point_details(db, trace_id=trace)
         assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
@@ -48,8 +50,8 @@ def test_point_scores_by_trace() -> None:
         db.close()
 
 
-def test_skipped_detail_does_not_create_a_score() -> None:
-    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
+def test_skipped_detail_creates_zero_score() -> None:
+    """仅有 skipped 时也应显示 0/1，而不是把计划券静默成空值。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -63,7 +65,7 @@ def test_skipped_detail_does_not_create_a_score() -> None:
         db.flush()
 
         scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert trace not in scores
+        assert scores[trace] == {"succeeded": 0, "tried": 1}
         assert "missing-trace" not in scores
         assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
     finally:
@@ -114,6 +116,126 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
         db.close()
 
 
+def test_report_uses_planned_count_when_events_are_partial_or_missing() -> None:
+    """退出前只有部分/没有逐券终态时，计划数仍是稳定分母。"""
+    db = SessionLocal()
+    report_date = date(2020, 1, 6)
+    partial_trace = "point-score-planned-partial"
+    empty_trace = "point-score-planned-empty"
+    try:
+        db.add_all([
+            CouponSession(
+                trace_id=partial_trace,
+                device_id="planned-partial-device",
+                status="abandoned",
+                app_env="prod",
+                platforms=[],
+                planned_coupon_count=8,
+                started_at=datetime(2020, 1, 6, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id=empty_trace,
+                device_id="planned-empty-device",
+                status="abandoned",
+                app_env="prod",
+                platforms=[],
+                planned_coupon_count=8,
+                started_at=datetime(2020, 1, 6, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponClaimEvent(
+                trace_id=partial_trace,
+                device_id="planned-partial-device",
+                coupon_id="mt-planned-success",
+                claim_date=report_date,
+                status="success",
+            ),
+        ])
+        db.flush()
+
+        report = coupon_data_report(
+            db,
+            date_from=report_date.isoformat(),
+            date_to=report_date.isoformat(),
+            app_env="prod",
+        )
+        rows = {item["trace_id"]: item for item in report["items"]}
+        assert rows[partial_trace]["point_success_count"] == 1
+        assert rows[partial_trace]["point_total_count"] == 8
+        assert rows[empty_trace]["point_success_count"] == 0
+        assert rows[empty_trace]["point_total_count"] == 8
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_user_records_include_actual_successes_with_planned_count() -> None:
+    """用户领券记录也必须加载逐券统计，不能把有成功事件的场次显示成 0/N。"""
+    db = SessionLocal()
+    trace = "point-score-user-records"
+    user_id = 987656
+    report_date = date(2020, 1, 8)
+    try:
+        db.add_all([
+            CouponSession(
+                trace_id=trace,
+                device_id="planned-user-records-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=[],
+                planned_coupon_count=8,
+                started_at=datetime(2020, 1, 8, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponClaimEvent(
+                trace_id=trace,
+                device_id="planned-user-records-device",
+                coupon_id="mt-user-records-success",
+                claim_date=report_date,
+                status="success",
+            ),
+        ])
+        db.flush()
+
+        records = coupon_user_records(db, user_id=user_id)
+        row = next(item for item in records["items"] if item["trace_id"] == trace)
+        assert row["point_success_count"] == 1
+        assert row["point_total_count"] == 8
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_planned_count_only_grows_for_a_trace() -> None:
+    """乱序帧不得用较小的队列总数覆盖已经观测到的计划数。"""
+    db = SessionLocal()
+    trace = "point-score-planned-upsert"
+    try:
+        row = CouponSession(
+            trace_id=trace,
+            device_id="planned-upsert-device",
+            status="started",
+            app_env="prod",
+            platforms=[],
+            started_at=datetime(2020, 1, 7, tzinfo=UTC),
+            started_date=date(2020, 1, 7),
+        )
+        db.add(row)
+        db.commit()
+
+        merge_session_planned_count(db, trace, 8)
+        merge_session_planned_count(db, trace, 3)
+        db.refresh(row)
+        assert row.planned_coupon_count == 8
+    finally:
+        db.rollback()
+        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
+        db.commit()
+        db.close()
+
+
 def test_coupon_point_details_endpoint() -> None:
     """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
     db = SessionLocal()