修复：收敛领券计划数迁移链

记录每轮计划券数，将跳过券计入逐场分母，并在中途退出或无逐券结果时返回稳定的成功率口径。

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: comparison_record_fail_reason
+"""
+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 = "comparison_record_fail_reason"
+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/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()

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/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/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,

tests/test_coupon_point_score.py

@@ -14,10 +14,11 @@ from app.admin.repositories.coupon_data import (
 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 +38,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 +49,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 +64,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 +115,88 @@ 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_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()