feat(coupon-data): 领券成功率看板(整单/点位/分平台 + 按券)

- coupon_session.platform_success:整单成功率②/点位成功率③ + 分平台点位成功率
- coupon_claim_record.app_env:按 coupon_id「按券成功率」聚合 + 子端点 /admin/api/coupon-data/coupons
- /step 复用 record_claims 同一 SessionLocal 写入(fire-and-forget),pricebot 转发/负载不变
- 迁移 coupon_session_platform_success + coupon_claim_app_env;新增 14 测试
- 设计见 docs/guides/领券成功率指标-设计与埋点.md

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/admin/repositories/coupon_data.py

@@ -5,17 +5,20 @@
 - 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
 - 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
 - summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
+- 另含 coupon_slot_report(数据源 coupon_claim_record):按 coupon_id「按券成功率」表,见设计 §13。
 """
 from __future__ import annotations
 
-from datetime import UTC, date as _date, datetime
+from datetime import UTC, datetime
+from datetime import date as _date
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import case, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
-from app.models.coupon_state import CouponSession
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 from app.models.user import User
+from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
 
 
 def _cn_hour(dt: datetime) -> int:
@@ -42,6 +45,46 @@ def _avg(vals: list[int]) -> int | None:
     return round(sum(vals) / len(vals)) if vals else None
 
 
+def _success_rates(rows: list) -> dict:
+    """平台粒度成功率(见 docs/guides/领券成功率指标-设计与埋点.md §3/§12):
+
+    - sel(s) = 勾选平台(`platforms` 空 → 全领三档 DEFAULT_PLATFORMS);
+    - succ(s) = `platform_success` ∩ sel(至少领到一张的平台);
+    - ② 整单成功率 = #{sel⊆succ 且 sel≠∅} / 发起数;
+    - ③ 点位成功率 = Σ|succ| / Σ|sel|;per_platform[p] = 勾了 p 且成功 / 勾了 p。
+    基数含全部 session(started/completed/failed/abandoned),与「发起数」同基数。
+    """
+    started = len(rows)
+    full_success = 0
+    point_success = 0
+    point_total = 0
+    per_succ = {p: 0 for p in DEFAULT_PLATFORMS}
+    per_total = {p: 0 for p in DEFAULT_PLATFORMS}
+    for r in rows:
+        sel = set(r.platforms) if r.platforms else set(DEFAULT_PLATFORMS)
+        succ = set(r.platform_success or []) & sel
+        point_success += len(succ)
+        point_total += len(sel)
+        if sel and succ == sel:
+            full_success += 1
+        for p in sel:
+            if p in per_total:  # 只统计三档已知平台;未知/非法平台 id 不进 per_platform
+                per_total[p] += 1
+                if p in succ:
+                    per_succ[p] += 1
+    return {
+        "full_success_count": full_success,
+        "full_success_rate": round(full_success / started, 4) if started else None,
+        "point_success_count": point_success,
+        "point_total_count": point_total,
+        "point_success_rate": round(point_success / point_total, 4) if point_total else None,
+        "per_platform": {
+            p: (round(per_succ[p] / per_total[p], 4) if per_total[p] else None)
+            for p in DEFAULT_PLATFORMS
+        },
+    }
+
+
 def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
     return {
@@ -131,6 +174,7 @@ def coupon_data_report(
         "p50_ms": _percentile(completed_elapsed, 50),
         "p95_ms": _percentile(completed_elapsed, 95),
         "p99_ms": _percentile(completed_elapsed, 99),
+        **_success_rates(rows),
     }
 
     # ── 按天趋势(柱=发起/完成数,线=平均耗时)──
@@ -223,3 +267,52 @@ 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()
     return {"items": [_session_to_row(r) for r in rows], "total": int(total)}
+
+
+_SLOT_OK = ("success", "already_claimed")
+_SLOT_TRIED = ("success", "already_claimed", "failed")
+
+
+def coupon_slot_report(
+    db: Session, *, date_from: str, date_to: str, app_env: str | None = None
+) -> dict:
+    """按 coupon_id(具体券)聚合成功率(见 docs/guides/领券成功率指标-设计与埋点.md §13)。
+
+    数据源 coupon_claim_record(粒度=设备-天,唯一键 device+coupon+day)。
+    - 尝试 = status ∈ {success, already_claimed, failed}(skipped 排除);
+    - 成功 = status ∈ {success, already_claimed};成功率 = 成功/尝试;
+    - claim_date 区间 + app_env(None=全部)过滤;按 tried 倒序返回。
+    """
+    d_from = _date.fromisoformat(date_from)
+    d_to = _date.fromisoformat(date_to)
+    ok = case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0)
+    stmt = (
+        select(
+            CouponClaimRecord.coupon_id,
+            func.max(CouponClaimRecord.coupon_name).label("coupon_name"),
+            func.count().label("tried"),
+            func.sum(ok).label("succeeded"),
+        )
+        .where(
+            CouponClaimRecord.claim_date >= d_from,
+            CouponClaimRecord.claim_date <= d_to,
+            CouponClaimRecord.status.in_(_SLOT_TRIED),
+        )
+        .group_by(CouponClaimRecord.coupon_id)
+    )
+    if app_env is not None:
+        stmt = stmt.where(CouponClaimRecord.app_env == app_env)
+    items = []
+    for coupon_id, coupon_name, tried, succeeded in db.execute(stmt).all():
+        tried = int(tried or 0)
+        succeeded = int(succeeded or 0)
+        items.append({
+            "coupon_id": coupon_id,
+            "coupon_name": coupon_name,
+            "platform": coupon_id_to_platform(coupon_id),
+            "tried": tried,
+            "succeeded": succeeded,
+            "success_rate": round(succeeded / tried, 4) if tried else None,
+        })
+    items.sort(key=lambda x: (-x["tried"], x["coupon_id"]))
+    return {"items": items}

app/admin/routers/coupon_data.py

@@ -18,6 +18,8 @@ from app.admin.schemas.coupon_data import (
     CouponDataOut,
     CouponDataRow,
     CouponDataSummary,
+    CouponSlotRow,
+    CouponSlotsOut,
     CouponUserRecordsOut,
 )
 from app.core.rewards import cn_today
@@ -87,6 +89,35 @@ def get_coupon_data(
     )
 
 
+@router.get(
+    "/coupons",
+    response_model=CouponSlotsOut,
+    summary="按券成功率(coupon_id 粒度;成功/(成功+失败),skipped 排除,设备-天口径)",
+)
+def get_coupon_slots(
+    db: AdminDb,
+    date_from: Annotated[str | None, Query(description="起始日 北京 YYYY-MM-DD,默认今天")] = None,
+    date_to: Annotated[str | None, Query(description="结束日 北京 YYYY-MM-DD,闭区间,默认=date_from")] = None,
+    app_env: Annotated[str, Query(description="prod(默认) / dev / all(全部环境)")] = "prod",
+) -> CouponSlotsOut:
+    today = cn_today()
+    d_from = _parse_day(date_from, field="date_from", default=today)
+    d_to = _parse_day(date_to, field="date_to", default=d_from)
+    if d_to < d_from:
+        raise HTTPException(status_code=422, detail="date_to 不能早于 date_from")
+    if (d_to - d_from).days + 1 > _MAX_RANGE_DAYS:
+        raise HTTPException(status_code=422, detail=f"区间最长 {_MAX_RANGE_DAYS} 天")
+    env = None if app_env == "all" else app_env
+    result = coupon_data.coupon_slot_report(
+        db, date_from=d_from.isoformat(), date_to=d_to.isoformat(), app_env=env
+    )
+    return CouponSlotsOut(
+        date_from=d_from.isoformat(),
+        date_to=d_to.isoformat(),
+        items=[CouponSlotRow(**r) for r in result["items"]],
+    )
+
+
 @router.get(
     "/user-records",
     response_model=CouponUserRecordsOut,

app/admin/schemas/coupon_data.py

@@ -19,6 +19,16 @@ class CouponDataSummary(BaseModel):
     p50_ms: int | None = Field(None, description="耗时 50 分位(ms,中位数)")
     p95_ms: int | None = Field(None, description="耗时 95 分位(ms)")
     p99_ms: int | None = Field(None, description="耗时 99 分位(ms)")
+    # 平台粒度成功率(见 docs/guides/领券成功率指标-设计与埋点.md):基数含全部 session。
+    full_success_count: int = Field(0, description="整单成功数(勾选平台全部领到的 session 数)")
+    full_success_rate: float | None = Field(None, description="整单成功率②=整单成功数/发起数;无数据为空")
+    point_success_count: int = Field(0, description="成功平台点位数(Σ 每次成功的平台数)")
+    point_total_count: int = Field(0, description="总平台点位数(Σ 每次勾选平台数;空勾选=全领三档)")
+    point_success_rate: float | None = Field(None, description="点位成功率③=成功点位/总点位;无数据为空")
+    per_platform: dict[str, float | None] = Field(
+        default_factory=dict,
+        description="分平台点位成功率 {平台id: rate|None};恒含美团/淘宝/京东三档,区间内无人勾选的平台为 None",
+    )
 
 
 class CouponDataDaily(BaseModel):
@@ -81,3 +91,22 @@ class CouponUserRecordsOut(BaseModel):
 
     items: list[CouponDataRow]
     total: int
+
+
+class CouponSlotRow(BaseModel):
+    """按券成功率一行(§13):粒度=设备-天;成功率=成功/(成功+失败),skipped 排除。"""
+
+    coupon_id: str
+    coupon_name: str | None = None
+    platform: str | None = Field(None, description="美团/淘宝/京东 平台 id;无法识别为空")
+    tried: int = Field(..., description="尝试数(success+already_claimed+failed 的设备-天数)")
+    succeeded: int = Field(..., description="成功数(success+already_claimed)")
+    success_rate: float | None = Field(None, description="成功率=成功/尝试")
+
+
+class CouponSlotsOut(BaseModel):
+    """按券成功率表响应(§13)。"""
+
+    date_from: str
+    date_to: str
+    items: list[CouponSlotRow]

app/api/v1/coupon.py

@@ -81,7 +81,16 @@ def _record_claims_blocking(
     device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
 ) -> None:
     with SessionLocal() as db:
-        coupon_repo.record_claims(db, device_id, user_id, trace_id, results)
+        # 取本次 session 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)。
+        app_env = coupon_repo.session_app_env(db, trace_id)
+        coupon_repo.record_claims(db, device_id, user_id, trace_id, results, app_env=app_env)
+        # 顺带把本帧「成功平台」并入 coupon_session.platform_success(admin 领券数据 ②整单/③点位成功率;
+        # 设计 route B,见 docs/guides/领券成功率指标-设计与埋点.md)。复用同一 SessionLocal、紧接 record_claims,
+        # 不新增连接;并集幂等(无新平台不写),trace_id 缺失或 session 行未落库则跳过。
+        if trace_id:
+            coupon_repo.merge_session_platform_success(
+                db, trace_id, coupon_repo.succeeded_platforms(results)
+            )
 
 
 def _mark_completed_blocking(

app/models/coupon_state.py

@@ -66,6 +66,9 @@ class CouponClaimRecord(Base):
 
     # success / already_claimed / failed / skipped(原样取 pricebot coupon 结果)
     status: Mapped[str] = mapped_column(String(24), nullable=False)
+    # 领券所属 session 的环境 prod/dev(/step 按 trace_id 查 coupon_session.app_env 打标)。
+    # 旧行 NULL(不回填)。admin「按券成功率」表据此过滤环境。见设计 §13。
+    app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
     vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
     coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
     # 这张领到几张(pricebot display_count;给不出时为 None)
@@ -240,6 +243,10 @@ class CouponSession(Base):
     platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
     # 领到总张数(收尾帧带)。
     claimed_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 → 视作空集。
+    platform_success: Mapped[list | None] = mapped_column(_JSON, nullable=True)
     # pricebot done 帧回传的公网调试链接(price.shaguabijia.com/traces/{dir});含落盘时分秒、拼不出,只能存
     # (同 ComparisonRecord.trace_url)。admin「领券数据」明细据此渲染可点 trace 链接;未到 done(failed/abandoned)为空。
     trace_url: Mapped[str | None] = mapped_column(String(512), nullable=True)

app/repositories/coupon_state.py

@@ -147,12 +147,22 @@ def reset_today_completion(db: Session, device_id: str) -> int:
 
 # ===== 领券记录(coupon_claim_record)=====
 
+def session_app_env(db: Session, trace_id: str | None) -> str | None:
+    """按 trace_id 取 coupon_session.app_env(每券成功率表打环境标用);无 trace_id / 查不到 → None。"""
+    if not trace_id:
+        return None
+    return db.execute(
+        select(CouponSession.app_env).where(CouponSession.trace_id == trace_id)
+    ).scalar_one_or_none()
+
+
 def record_claims(
     db: Session,
     device_id: str,
     user_id: int | None,
     trace_id: str | None,
     results: list[dict],
+    app_env: str | None = None,
 ) -> int:
     """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。
 
@@ -186,12 +196,15 @@ def record_claims(
                 row.user_id = user_id
             if count is not None:
                 row.claimed_count = count
+            if app_env is not None:
+                row.app_env = app_env
             row.extra = r
         else:
             db.add(CouponClaimRecord(
                 device_id=device_id, user_id=user_id,
                 coupon_id=coupon_id, claim_date=today,
-                status=status, vendor=r.get("vendor"), coupon_name=r.get("name"),
+                status=status, app_env=app_env,
+                vendor=r.get("vendor"), coupon_name=r.get("name"),
                 claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
                 extra=r,
             ))
@@ -235,6 +248,47 @@ def sum_claimed_count(db: Session, user_id: int) -> int:
     return int(total or 0)
 
 
+# ===== 领券平台推导(coupon_id → 平台;成功平台集)=====
+
+# 成功语义:success + already_claimed 算成功(pricebot 代码 emit already_claimed,协议 enum 漏了);
+# failed / skipped 不算。与 sum_claimed_count 同口径。
+_SUCCESS_STATUSES = frozenset({"success", "already_claimed"})
+
+# 三档平台 id 及固定序(美团→淘宝→京东),与客户端 DEFAULT_PLATFORM_ORDER 对齐。
+DEFAULT_PLATFORMS: tuple[str, ...] = ("meituan-waimai", "taobao-shanguang", "jd-waimai")
+
+
+def coupon_id_to_platform(coupon_id: str | None) -> str | None:
+    """coupon_id 前缀 → 平台 id;无法识别 / 空 → None。
+
+    与客户端 `CouponForegroundService.couponIdToPlatform` 同词表:
+    mt_→美团外卖 / tb_·ele_·elm_→淘宝闪购 / jd_→京东外卖。
+    """
+    if not coupon_id:
+        return None
+    if coupon_id.startswith("mt_"):
+        return "meituan-waimai"
+    if coupon_id.startswith(("tb_", "ele_", "elm_")):
+        return "taobao-shanguang"
+    if coupon_id.startswith("jd_"):
+        return "jd-waimai"
+    return None
+
+
+def succeeded_platforms(results: list[dict]) -> list[str]:
+    """一批券结果 → 至少领到一张的平台集(按 DEFAULT_PLATFORMS 去重保序)。
+
+    只取 status∈{success, already_claimed} 的券;失败/跳过、无法识别平台的券跳过。
+    """
+    ok: set[str] = set()
+    for r in results:
+        if r.get("status") in _SUCCESS_STATUSES:
+            platform = coupon_id_to_platform(r.get("coupon_id"))
+            if platform is not None:
+                ok.add(platform)
+    return [p for p in DEFAULT_PLATFORMS if p in ok]
+
+
 # ===== 领券任务流水(coupon_session,admin「领券数据」看板数据源)=====
 
 def upsert_coupon_session(
@@ -321,3 +375,31 @@ def upsert_coupon_session(
     except IntegrityError:
         # 并发下另一请求刚插了同 trace_id → 唯一约束撞,回滚忽略(本就幂等)。
         db.rollback()
+
+
+def merge_session_platform_success(
+    db: Session, trace_id: str, platforms: list[str]
+) -> None:
+    """把本帧「成功平台」并入 coupon_session.platform_success(按 trace_id,并集幂等,按 DEFAULT_PLATFORMS 保序)。
+
+    - 领券 /step 每逢带券结果的帧调一次(平台成败布尔,跨帧取并集天然幂等,不重复计)。
+    - 读不到该 trace_id 的行 → **静默跳过**(不建兜底行;设计 §5:started 帧几乎必先落库)。
+    - 并集无变化(该平台已记过)→ 不写库,省一次 UPDATE。
+    - fire-and-forget:调用方已吞异常;并发唯一冲突回滚忽略。
+    """
+    if not platforms:
+        return
+    row = db.execute(
+        select(CouponSession).where(CouponSession.trace_id == trace_id)
+    ).scalar_one_or_none()
+    if row is None:
+        return
+    merged = set(row.platform_success or []) | set(platforms)
+    new_list = [p for p in DEFAULT_PLATFORMS if p in merged]
+    if new_list == (row.platform_success or []):
+        return  # 幂等:无新平台,不写
+    row.platform_success = new_list
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()