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

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

alembic/versions/coupon_session_planned_count.py

@@ -1,28 +0,0 @@
-"""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,7 +23,6 @@ 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:
@@ -167,10 +166,6 @@ 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,
@@ -187,8 +182,8 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": succeeded if point_total > 0 else None,
-        "point_total_count": point_total if point_total > 0 else None,
+        "point_success_count": point_stats["succeeded"] if point_stats else None,
+        "point_total_count": point_stats["tried"] if point_stats else None,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
@@ -207,7 +202,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_(_SESSION_POINT_TOTAL),
+            CouponClaimEvent.status.in_(_SLOT_TRIED),
         )
         .group_by(CouponClaimEvent.trace_id)
     ).all()

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

@@ -93,13 +93,6 @@ 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:
@@ -183,18 +176,6 @@ 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,9 +277,6 @@ 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,22 +157,6 @@ 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/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))

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()

tests/test_coupon_point_score.py

@@ -14,11 +14,10 @@ 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:
@@ -38,7 +37,7 @@ def test_point_scores_by_trace() -> None:
 
         stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
-        assert stats["tried"] == 4
+        assert stats["tried"] == 3
         details = coupon_point_details(db, trace_id=trace)
         assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
@@ -49,8 +48,8 @@ def test_point_scores_by_trace() -> None:
         db.close()
 
 
-def test_skipped_detail_creates_zero_score() -> None:
-    """仅有 skipped 时也应显示 0/1，而不是把计划券静默成空值。"""
+def test_skipped_detail_does_not_create_a_score() -> None:
+    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -64,7 +63,7 @@ def test_skipped_detail_creates_zero_score() -> None:
         db.flush()
 
         scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert scores[trace] == {"succeeded": 0, "tried": 1}
+        assert trace not in scores
         assert "missing-trace" not in scores
         assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
     finally:
@@ -115,88 +114,6 @@ 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()