Merge origin/main into feat/ad-coin-ecpm-unit-audit (对齐最新 main,清除分支过期)

- eCPM 口径修正:getEcpm 原值是分/千次,新增 parse_ecpm_fen;parse_ecpm_yuan
  改为 ÷100 转元,因子1 阈值按元判档(100/200/400),收益换算用元
- test-grant 用客户端按 ad_session_id 上报的真实 eCPM 发奖(取不到/≤0 兜底 200)
- 新增 GET /admin/api/ad-coin-audit 只读复算对账(router/repo/schema),复用发奖公式
- 同步更新 docs/api、docs/database 的分/元口径

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -79,6 +79,10 @@ WXPAY_MCH_PRIVATE_KEY_PATH=./secrets/apiclient_key.pem
 WXPAY_PUBLIC_KEY_ID=PUB_KEY_ID_xxxxxxxx
 WXPAY_PUBLIC_KEY_PATH=./secrets/pub_key.pem
 WXPAY_TRANSFER_SCENE_ID=1000
+WXPAY_AUTH_NOTIFY_URL=
+WITHDRAW_AUTO_RECONCILE_ENABLED=false
+WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC=300
+WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES=15
 
 # ===== 穿山甲激励视频(服务端发奖回调)=====
 # 看完激励视频后穿山甲服务器 S2S 回调本服务发金币(客户端不参与发奖)。

.gitignore

@@ -43,3 +43,4 @@ secrets/*
 
 # 运行日志(run.sh 输出, 不入库)
 *.log
+logs/

alembic/versions/ad_feed_reward_session_id.py

@@ -0,0 +1,38 @@
+"""add ad_session_id to ad feed reward record
+
+Revision ID: ad_feed_reward_session
+Revises: coin_reward_phase2
+Create Date: 2026-06-09
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "ad_feed_reward_session"
+down_revision = "coin_reward_phase2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ad_feed_reward_record",
+        sa.Column("ad_session_id", sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_feed_reward_record_ad_session_id"),
+        "ad_feed_reward_record",
+        ["ad_session_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f("ix_ad_feed_reward_record_ad_session_id"),
+        table_name="ad_feed_reward_record",
+    )
+    op.drop_column("ad_feed_reward_record", "ad_session_id")

alembic/versions/coin_reward_points_phase2.py

@@ -0,0 +1,71 @@
+"""coin reward points phase2
+
+Revision ID: coin_reward_phase2
+Revises: 0cf18d590b1d
+Create Date: 2026-06-08
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "coin_reward_phase2"
+down_revision = "0cf18d590b1d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ad_ecpm_record",
+        sa.Column("ad_session_id", sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_ecpm_record_ad_session_id"),
+        "ad_ecpm_record",
+        ["ad_session_id"],
+        unique=False,
+    )
+    op.create_index(
+        "uq_ad_ecpm_record_session",
+        "ad_ecpm_record",
+        ["ad_session_id"],
+        unique=True,
+    )
+
+    op.add_column(
+        "ad_reward_record",
+        sa.Column(
+            "reward_scene",
+            sa.String(length=32),
+            nullable=False,
+            server_default="reward_video",
+        ),
+    )
+    op.add_column(
+        "ad_reward_record",
+        sa.Column("ad_session_id", sa.String(length=64), nullable=True),
+    )
+    op.add_column(
+        "ad_reward_record",
+        sa.Column("ecpm_raw", sa.String(length=32), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_reward_record_ad_session_id"),
+        "ad_reward_record",
+        ["ad_session_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_ad_reward_record_ad_session_id"), table_name="ad_reward_record")
+    op.drop_column("ad_reward_record", "ecpm_raw")
+    op.drop_column("ad_reward_record", "ad_session_id")
+    op.drop_column("ad_reward_record", "reward_scene")
+
+    op.drop_index("uq_ad_ecpm_record_session", table_name="ad_ecpm_record")
+    op.drop_index(op.f("ix_ad_ecpm_record_ad_session_id"), table_name="ad_ecpm_record")
+    op.drop_column("ad_ecpm_record", "ad_session_id")

alembic/versions/merge_ad_feed_reward_session_and_withdraw_safety_heads.py

@@ -0,0 +1,26 @@
+"""merge ad feed reward session and withdraw safety heads
+
+Revision ID: 044dce6e9b1f
+Revises: ad_feed_reward_session, withdraw_safety_indexes
+Create Date: 2026-06-09 09:52:08.276767
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '044dce6e9b1f'
+down_revision: Union[str, Sequence[str], None] = ('ad_feed_reward_session', 'withdraw_safety_indexes')
+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/withdraw_safety_indexes.py

@@ -0,0 +1,42 @@
+"""add withdraw concurrency safety indexes
+
+Revision ID: withdraw_safety_indexes
+Revises: 0cf18d590b1d
+Create Date: 2026-06-08 16:20:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "withdraw_safety_indexes"
+down_revision: Union[str, Sequence[str], None] = "0cf18d590b1d"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ux_withdraw_order_user_active",
+        "withdraw_order",
+        ["user_id"],
+        unique=True,
+        sqlite_where=sa.text("status IN ('reviewing', 'pending')"),
+        postgresql_where=sa.text("status IN ('reviewing', 'pending')"),
+    )
+    op.create_index(
+        "ux_cash_transaction_withdraw_refund_ref",
+        "cash_transaction",
+        ["ref_id"],
+        unique=True,
+        sqlite_where=sa.text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"),
+        postgresql_where=sa.text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ux_cash_transaction_withdraw_refund_ref", table_name="cash_transaction")
+    op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order")

app/admin/main.py

@@ -13,6 +13,7 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from app.admin.routers.ad_audit import router as ad_audit_router
 from app.admin.routers.admins import router as admins_router
 from app.admin.routers.audit import router as audit_router
 from app.admin.routers.auth import router as auth_router
@@ -84,3 +85,4 @@ admin_app.include_router(feedback_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(audit_router)
 admin_app.include_router(config_router)
+admin_app.include_router(ad_audit_router)

app/admin/repositories/ad_audit.py

@@ -0,0 +1,168 @@
+"""看广告金币审计:复算 expected_coin 并与实发对比。
+
+只读。复用 [app.core.rewards] 的公式函数(不另写公式,避免与正式发奖口径漂移):
+- 看视频:每条 granted = 1 份,第 N 份 = 当日该用户 granted 的 reward_video 顺序号
+  (与 ad_reward.grant_ad_reward 里 `_granted_today + 1` 一致)。
+- 信息流:每条按 unit_count 份逐份累加,LT 序号 = 当日该用户已 granted 份数累计
+  (与 ad_feed_reward._unit_reward_total 的 existing_units 一致)。
+
+非 granted(capped/ecpm_missing)不占用份序号、应发恒 0,据此校验闸口是否确实没发。
+"""
+from __future__ import annotations
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.core import rewards
+from app.models.ad_feed_reward import AdFeedRewardRecord
+from app.models.ad_reward import AdRewardRecord
+from app.repositories.ad_feed_reward import FEED_REWARD_UNIT_SECONDS
+
+
+def _reward_video_rows(
+    db: Session, *, date: str, user_id: int | None
+) -> list[dict]:
+    """看视频记录复算。按 (user_id, created_at) 升序还原当日第 N 份。"""
+    stmt = (
+        select(AdRewardRecord)
+        .where(
+            AdRewardRecord.reward_date == date,
+            AdRewardRecord.reward_scene == "reward_video",
+        )
+        .order_by(AdRewardRecord.user_id, AdRewardRecord.created_at)
+    )
+    if user_id is not None:
+        stmt = stmt.where(AdRewardRecord.user_id == user_id)
+
+    granted_n: dict[int, int] = {}  # user_id -> 已 granted 份数
+    rows: list[dict] = []
+    for rec in db.execute(stmt).scalars():
+        if rec.status == "granted":
+            nth = granted_n.get(rec.user_id, 0) + 1
+            granted_n[rec.user_id] = nth
+            expected = rewards.calculate_ad_reward_coin(rec.ecpm_raw, nth)
+            rows.append({
+                "scene": "reward_video",
+                "record_id": rec.id,
+                "user_id": rec.user_id,
+                "created_at": rec.created_at,
+                "status": rec.status,
+                "ecpm": rec.ecpm_raw,
+                "ecpm_factor": rewards.ad_ecpm_factor(rewards.parse_ecpm_yuan(rec.ecpm_raw)),
+                "units": 1,
+                "lt_index_start": nth,
+                "lt_index_end": nth,
+                "lt_factor_start": rewards.ad_lt_factor(nth),
+                "lt_factor_end": rewards.ad_lt_factor(nth),
+                "expected_coin": expected,
+                "actual_coin": rec.coin,
+                "matched": expected == rec.coin,
+            })
+        else:
+            # capped / ecpm_missing:不发金币,校验实发确为 0
+            rows.append({
+                "scene": "reward_video",
+                "record_id": rec.id,
+                "user_id": rec.user_id,
+                "created_at": rec.created_at,
+                "status": rec.status,
+                "ecpm": rec.ecpm_raw,
+                "ecpm_factor": None,
+                "units": 1,
+                "lt_index_start": None,
+                "lt_index_end": None,
+                "lt_factor_start": None,
+                "lt_factor_end": None,
+                "expected_coin": 0,
+                "actual_coin": rec.coin,
+                "matched": rec.coin == 0,
+            })
+    return rows
+
+
+def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
+    """信息流记录复算。granted 记录逐份累加,LT 序号沿用当日累计份数。"""
+    stmt = (
+        select(AdFeedRewardRecord)
+        .where(AdFeedRewardRecord.reward_date == date)
+        .order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at)
+    )
+    if user_id is not None:
+        stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
+
+    granted_units: dict[int, int] = {}  # user_id -> 已 granted 份数累计
+    rows: list[dict] = []
+    for rec in db.execute(stmt).scalars():
+        if rec.status == "granted":
+            existing = granted_units.get(rec.user_id, 0)
+            units = rec.unit_count
+            expected = sum(
+                rewards.calculate_ad_reward_coin(rec.ecpm_raw, existing + offset)
+                for offset in range(1, units + 1)
+            )
+            granted_units[rec.user_id] = existing + units
+            start = existing + 1 if units > 0 else None
+            end = existing + units if units > 0 else None
+            rows.append({
+                "scene": "feed",
+                "record_id": rec.id,
+                "user_id": rec.user_id,
+                "created_at": rec.created_at,
+                "status": rec.status,
+                "ecpm": rec.ecpm_raw,
+                "ecpm_factor": rewards.ad_ecpm_factor(rewards.parse_ecpm_yuan(rec.ecpm_raw)),
+                "units": units,
+                "lt_index_start": start,
+                "lt_index_end": end,
+                "lt_factor_start": rewards.ad_lt_factor(start) if start else None,
+                "lt_factor_end": rewards.ad_lt_factor(end) if end else None,
+                "expected_coin": expected,
+                "actual_coin": rec.coin,
+                "matched": expected == rec.coin,
+            })
+        else:
+            rows.append({
+                "scene": "feed",
+                "record_id": rec.id,
+                "user_id": rec.user_id,
+                "created_at": rec.created_at,
+                "status": rec.status,
+                "ecpm": rec.ecpm_raw,
+                "ecpm_factor": None,
+                "units": rec.unit_count,
+                "lt_index_start": None,
+                "lt_index_end": None,
+                "lt_factor_start": None,
+                "lt_factor_end": None,
+                "expected_coin": 0,
+                "actual_coin": rec.coin,
+                "matched": rec.coin == 0,
+            })
+    return rows
+
+
+def ad_coin_audit(
+    db: Session, *, date: str, user_id: int | None, scene: str | None, limit: int
+) -> list[dict]:
+    """返回当日发奖复算明细,按 created_at 倒序(最新在前)截断到 limit。
+
+    scene: None=两类都要 / "reward_video" / "feed"。
+    份序号在截断前已基于全天数据算好,故 limit 只影响展示条数、不影响 expected 复算正确性。
+    """
+    rows: list[dict] = []
+    if scene in (None, "reward_video"):
+        rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
+    if scene in (None, "feed"):
+        rows.extend(_feed_rows(db, date=date, user_id=user_id))
+    rows.sort(key=lambda r: r["created_at"], reverse=True)
+    return rows[:limit]
+
+
+def formula_snapshot() -> dict:
+    """当前公式参数快照(给前端展示规则参照)。直接读 rewards 常量,与发奖同源。"""
+    return {
+        "coin_per_yuan": rewards.COIN_PER_YUAN,
+        "feed_unit_seconds": FEED_REWARD_UNIT_SECONDS,
+        "ecpm_factor_tiers": [list(t) for t in rewards.AD_ECPM_FACTOR_TABLE],
+        "lt_factor_tiers": [list(t) for t in rewards.AD_LT_FACTOR_TABLE],
+    }

app/admin/repositories/queries.py

@@ -5,9 +5,13 @@
 """
 from __future__ import annotations
 
-from sqlalchemy import Select, func, select
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import Select, asc, desc, func, or_, select
 from sqlalchemy.orm import Session
 
+from app.models.admin import AdminAuditLog
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
 from app.models.user import User
@@ -88,15 +92,114 @@ def list_all_withdraw_orders(
     *,
     user_id: int | None = None,
     status: str | None = None,
+    keyword: str | None = None,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+    date_field: str = "created_at",
+    sort_by: str = "created_at",
+    sort_order: str = "desc",
+    quick_filter: str | None = None,
     limit: int = 20,
     cursor: int | None = None,
 ) -> tuple[list[WithdrawOrder], int | None]:
     stmt = select(WithdrawOrder)
+    needs_user_join = bool(keyword and keyword.strip()) or quick_filter == "high_risk"
+    if needs_user_join:
+        stmt = stmt.outerjoin(User, User.id == WithdrawOrder.user_id)
+
     if user_id is not None:
         stmt = stmt.where(WithdrawOrder.user_id == user_id)
     if status:
         stmt = stmt.where(WithdrawOrder.status == status)
-    return cursor_paginate(db, stmt, WithdrawOrder.id, limit=limit, cursor=cursor)
+
+    kw = (keyword or "").strip()
+    if kw:
+        pattern = f"%{kw}%"
+        conditions = [
+            WithdrawOrder.out_bill_no.ilike(pattern),
+            WithdrawOrder.transfer_bill_no.ilike(pattern),
+            WithdrawOrder.user_name.ilike(pattern),
+            WithdrawOrder.wechat_state.ilike(pattern),
+            WithdrawOrder.fail_reason.ilike(pattern),
+            User.phone.ilike(pattern),
+            User.nickname.ilike(pattern),
+            User.wechat_nickname.ilike(pattern),
+        ]
+        if kw.isdigit():
+            conditions.append(WithdrawOrder.user_id == int(kw))
+        stmt = stmt.where(or_(*conditions))
+
+    date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at
+    if date_from is not None:
+        stmt = stmt.where(date_col >= _as_utc_naive(date_from))
+    if date_to is not None:
+        stmt = stmt.where(date_col <= _as_utc_naive(date_to))
+
+    now = datetime.now(timezone.utc).replace(tzinfo=None)
+    today_start = (
+        datetime.now(ZoneInfo("Asia/Shanghai"))
+        .replace(hour=0, minute=0, second=0, microsecond=0)
+        .astimezone(timezone.utc)
+        .replace(tzinfo=None)
+    )
+    if quick_filter == "abnormal":
+        stmt = stmt.where(
+            or_(
+                WithdrawOrder.status.in_(("failed", "rejected")),
+                WithdrawOrder.fail_reason.is_not(None),
+                WithdrawOrder.wechat_state.in_(("FAIL", "CANCELLED", "CLOSED")),
+            )
+        )
+    elif quick_filter == "failed":
+        stmt = stmt.where(WithdrawOrder.status == "failed")
+    elif quick_filter == "overdue_reviewing":
+        stmt = stmt.where(
+            WithdrawOrder.status == "reviewing",
+            WithdrawOrder.created_at <= now - timedelta(minutes=30),
+        )
+    elif quick_filter == "pending_overdue":
+        stmt = stmt.where(
+            WithdrawOrder.status == "pending",
+            WithdrawOrder.updated_at <= now - timedelta(minutes=15),
+        )
+    elif quick_filter == "today":
+        stmt = stmt.where(WithdrawOrder.created_at >= today_start)
+    elif quick_filter == "high_risk":
+        stmt = stmt.where(
+            or_(
+                WithdrawOrder.user_name.is_(None),
+                WithdrawOrder.user_name == "",
+                User.status != "active",
+                User.created_at >= now - timedelta(hours=24),
+                WithdrawOrder.status.in_(("failed", "rejected")),
+                WithdrawOrder.fail_reason.is_not(None),
+            )
+        )
+
+    sort_cols = {
+        "id": WithdrawOrder.id,
+        "created_at": WithdrawOrder.created_at,
+        "updated_at": WithdrawOrder.updated_at,
+        "amount_cents": WithdrawOrder.amount_cents,
+    }
+    sort_col = sort_cols.get(sort_by, WithdrawOrder.created_at)
+    order_fn = asc if sort_order == "asc" else desc
+    id_order = asc(WithdrawOrder.id) if sort_order == "asc" else desc(WithdrawOrder.id)
+    stmt = stmt.order_by(order_fn(sort_col), id_order)
+
+    offset = max(cursor or 0, 0)
+    rows = list(db.execute(stmt.offset(offset).limit(limit + 1)).scalars().all())
+    has_more = len(rows) > limit
+    items = rows[:limit]
+    next_cursor = offset + limit if has_more else None
+    return items, next_cursor
+
+
+def _as_utc_naive(value: datetime) -> datetime:
+    """前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。"""
+    if value.tzinfo is None:
+        return value
+    return value.astimezone(timezone.utc).replace(tzinfo=None)
 
 
 def list_feedbacks(
@@ -122,6 +225,147 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder
     ).scalar_one_or_none()
 
 
+def withdraw_summary(db: Session) -> dict:
+    """提现审核台顶部统计。金额单位:分。"""
+    rows = db.execute(
+        select(
+            WithdrawOrder.status,
+            func.count(WithdrawOrder.id),
+            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+        ).group_by(WithdrawOrder.status)
+    ).all()
+    by_status = {
+        status: {"count": int(count), "amount_cents": int(amount_cents)}
+        for status, count, amount_cents in rows
+    }
+
+    today_start = (
+        datetime.now(ZoneInfo("Asia/Shanghai"))
+        .replace(hour=0, minute=0, second=0, microsecond=0)
+        .astimezone(timezone.utc)
+    )
+
+    def _today_count(status: str) -> int:
+        return db.execute(
+            select(func.count(WithdrawOrder.id)).where(
+                WithdrawOrder.status == status,
+                WithdrawOrder.updated_at >= today_start,
+            )
+        ).scalar_one()
+
+    today_success_amount = db.execute(
+        select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
+            WithdrawOrder.status == "success",
+            WithdrawOrder.updated_at >= today_start,
+        )
+    ).scalar_one()
+
+    return {
+        "reviewing_count": by_status.get("reviewing", {}).get("count", 0),
+        "reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
+        "pending_count": by_status.get("pending", {}).get("count", 0),
+        "failed_count": by_status.get("failed", {}).get("count", 0),
+        "today_success_count": _today_count("success"),
+        "today_success_amount_cents": int(today_success_amount),
+        "today_rejected_count": _today_count("rejected"),
+    }
+
+
+def list_withdraw_audit_logs(
+    db: Session, out_bill_no: str, *, limit: int = 20
+) -> list[AdminAuditLog]:
+    return list(
+        db.execute(
+            select(AdminAuditLog)
+            .where(AdminAuditLog.target_type == "withdraw", AdminAuditLog.target_id == out_bill_no)
+            .order_by(AdminAuditLog.id.desc())
+            .limit(limit)
+        ).scalars().all()
+    )
+
+
+def withdraw_risk_flags(
+    order: WithdrawOrder,
+    user: User | None,
+    recent_withdraws: list[WithdrawOrder],
+    cash_balance_cents: int,
+) -> tuple[list[str], int]:
+    flags: list[str] = []
+    if not order.user_name:
+        flags.append("缺少提现实名")
+    if user and user.status != "active":
+        flags.append(f"账号状态:{user.status}")
+    if user and user.created_at:
+        created_at = user.created_at.replace(tzinfo=timezone.utc) if user.created_at.tzinfo is None else user.created_at
+        if datetime.now(timezone.utc) - created_at < timedelta(hours=24):
+            flags.append("新注册用户")
+    failed_or_rejected = sum(1 for item in recent_withdraws if item.status in {"failed", "rejected"})
+    if failed_or_rejected:
+        flags.append(f"历史异常提现{failed_or_rejected}笔")
+    recent_reviewing = sum(1 for item in recent_withdraws if item.status == "reviewing")
+    if recent_reviewing >= 3:
+        flags.append(f"待审核提现偏多:{recent_reviewing}笔")
+    if cash_balance_cents < 0:
+        flags.append("现金余额为负")
+    score = min(100, len(flags) * 20 + failed_or_rejected * 10)
+    return flags, score
+
+
+def withdraw_ledger_check(db: Session) -> dict:
+    cash_balance_total = int(
+        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
+    )
+    cash_txn_total = int(
+        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
+    )
+
+    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
+    cash_txns = list(
+        db.execute(
+            select(CashTransaction).where(
+                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    withdraw_refs = {txn.ref_id for txn in cash_txns if txn.biz_type == "withdraw"}
+    refund_counts: dict[str, int] = {}
+    for txn in cash_txns:
+        if txn.biz_type == "withdraw_refund" and txn.ref_id:
+            refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
+
+    missing_withdraw = 0
+    missing_refund = 0
+    refund_on_non_terminal = 0
+    for order in orders:
+        if order.out_bill_no not in withdraw_refs:
+            missing_withdraw += 1
+        has_refund = refund_counts.get(order.out_bill_no, 0) > 0
+        if order.status in {"failed", "rejected"} and not has_refund:
+            missing_refund += 1
+        if has_refund and order.status not in {"failed", "rejected"}:
+            refund_on_non_terminal += 1
+
+    duplicate_refund = sum(1 for count in refund_counts.values() if count > 1)
+    diff = cash_balance_total - cash_txn_total
+    ok = (
+        diff == 0
+        and missing_withdraw == 0
+        and missing_refund == 0
+        and duplicate_refund == 0
+        and refund_on_non_terminal == 0
+    )
+    return {
+        "ok": ok,
+        "cash_balance_total_cents": cash_balance_total,
+        "cash_transaction_total_cents": cash_txn_total,
+        "balance_diff_cents": diff,
+        "missing_withdraw_txn_count": missing_withdraw,
+        "missing_refund_txn_count": missing_refund,
+        "duplicate_refund_txn_count": duplicate_refund,
+        "refund_txn_on_non_terminal_count": refund_on_non_terminal,
+    }
+
+
 def get_user_overview(db: Session, user_id: int) -> dict | None:
     """用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
     user = db.get(User, user_id)

app/admin/repositories/stats.py

@@ -10,8 +10,11 @@ from datetime import datetime, timedelta, timezone
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.models.ad_feed_reward import AdFeedRewardRecord
+from app.models.ad_reward import AdRewardRecord
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
+from app.models.signin import SigninBoostRecord, SigninRecord
 from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
@@ -71,6 +74,37 @@ def dashboard_overview(db: Session) -> dict:
         "coins": {
             # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
             "granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
+            "reward_video_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
+            ),
+            "reward_video_watch_count": _count(
+                AdRewardRecord,
+                AdRewardRecord.reward_scene == "reward_video",
+                AdRewardRecord.status == "granted",
+            ),
+            "feed_ad_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type == "feed_ad_reward",
+            ),
+            "feed_ad_watch_count": _count(
+                AdFeedRewardRecord,
+                AdFeedRewardRecord.status == "granted",
+            ),
+            "signin_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type == "signin",
+            ),
+            "signin_count": _count(SigninRecord),
+            "signin_boost_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type == "signin_boost",
+            ),
+            "signin_boost_watch_count": _count(SigninBoostRecord),
         },
         "cash": {
             "withdraw_success_cents": _sum(

app/admin/routers/ad_audit.py

@@ -0,0 +1,44 @@
+"""admin 看广告金币审计:只读对账,核对发奖金币是否按公式计算。
+
+任意已登录 admin 可看(只读,不涉及资金操作)。复算逻辑在 app/admin/repositories/ad_audit.py。
+"""
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query
+
+from app.admin.deps import AdminDb, get_current_admin
+from app.admin.repositories import ad_audit
+from app.admin.schemas.ad_audit import AdCoinAuditOut, AdCoinAuditRow, AdCoinFormulaOut
+from app.core.rewards import cn_today
+
+router = APIRouter(
+    prefix="/admin/api/ad-coin-audit",
+    tags=["admin-ad-coin-audit"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+
+@router.get("", response_model=AdCoinAuditOut, summary="看广告金币公式审计(复算对比)")
+def get_ad_coin_audit(
+    db: AdminDb,
+    date: Annotated[str | None, Query(description="北京时间 YYYY-MM-DD,默认今天")] = None,
+    user_id: Annotated[int | None, Query(description="只看某用户;不传=全部用户")] = None,
+    scene: Annotated[
+        str | None, Query(description="reward_video / feed;不传=两类都要")
+    ] = None,
+    limit: Annotated[int, Query(ge=1, le=500)] = 100,
+) -> AdCoinAuditOut:
+    audit_date = date or cn_today().isoformat()
+    rows = ad_audit.ad_coin_audit(
+        db, date=audit_date, user_id=user_id, scene=scene, limit=limit,
+    )
+    items = [AdCoinAuditRow(**r) for r in rows]
+    return AdCoinAuditOut(
+        date=audit_date,
+        formula=AdCoinFormulaOut(**ad_audit.formula_snapshot()),
+        total=len(items),
+        mismatch_count=sum(1 for it in items if not it.matched),
+        items=items,
+    )

app/admin/routers/withdraw.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
@@ -14,7 +15,23 @@ from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.repositories import queries
 from app.admin.schemas.common import CursorPage
-from app.admin.schemas.wallet import ReconcileResult, WithdrawOrderOut, WithdrawRejectRequest
+from app.admin.schemas.admin import AdminAuditLogOut
+from app.admin.schemas.wallet import (
+    CashTxnOut,
+    ReconcileResult,
+    WithdrawBulkRejectRequest,
+    WithdrawBulkRequest,
+    WithdrawBulkResult,
+    WithdrawBulkItemResult,
+    WithdrawDetailOut,
+    WithdrawLedgerCheckOut,
+    WithdrawOrderOut,
+    WithdrawRejectRequest,
+    WithdrawSummaryOut,
+    WithdrawUserSnapshot,
+    WxpayHealthCheckOut,
+)
+from app.core.config import settings
 from app.integrations import wxpay
 from app.models.admin import AdminUser
 from app.repositories import wallet as wallet_repo
@@ -31,17 +48,143 @@ def list_withdraws(
     db: AdminDb,
     user_id: Annotated[int | None, Query()] = None,
     status: Annotated[str | None, Query()] = None,
+    keyword: Annotated[str | None, Query(max_length=100)] = None,
+    date_from: Annotated[datetime | None, Query()] = None,
+    date_to: Annotated[datetime | None, Query()] = None,
+    date_field: Annotated[str, Query(pattern="^(created_at|updated_at)$")] = "created_at",
+    sort_by: Annotated[
+        str, Query(pattern="^(id|created_at|updated_at|amount_cents)$")
+    ] = "created_at",
+    sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
+    quick_filter: Annotated[
+        str | None,
+        Query(pattern="^(abnormal|failed|overdue_reviewing|pending_overdue|today|high_risk)$"),
+    ] = None,
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[WithdrawOrderOut]:
     items, next_cursor = queries.list_all_withdraw_orders(
-        db, user_id=user_id, status=status, limit=limit, cursor=cursor,
+        db,
+        user_id=user_id,
+        status=status,
+        keyword=keyword,
+        date_from=date_from,
+        date_to=date_to,
+        date_field=date_field,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        quick_filter=quick_filter,
+        limit=limit,
+        cursor=cursor,
     )
     return CursorPage(
         items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor,
     )
 
 
+@router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
+def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
+    return WithdrawSummaryOut(**queries.withdraw_summary(db))
+
+
+@router.get("/health-check", response_model=WxpayHealthCheckOut, summary="提现配置健康检查")
+def withdraw_health_check() -> WxpayHealthCheckOut:
+    private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH)  # noqa: SLF001
+    public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH)  # noqa: SLF001
+    issues: list[str] = []
+
+    private_loadable = False
+    public_loadable = False
+    try:
+        wxpay._load_private_key()  # noqa: SLF001
+        private_loadable = True
+    except Exception as e:  # noqa: BLE001
+        issues.append(f"商户私钥不可用:{e}")
+    try:
+        wxpay._load_public_key()  # noqa: SLF001
+        public_loadable = True
+    except Exception as e:  # noqa: BLE001
+        issues.append(f"微信支付平台公钥不可用:{e}")
+
+    if not settings.wxpay_configured:
+        issues.append("微信支付基础配置不完整")
+    if not settings.WXPAY_AUTH_NOTIFY_URL:
+        issues.append("免确认授权回调地址未配置")
+    if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
+        issues.append("自动对账未开启")
+
+    return WxpayHealthCheckOut(
+        ok=not issues,
+        wxpay_configured=settings.wxpay_configured,
+        wxpay_auth_configured=settings.wxpay_auth_configured,
+        private_key_path=str(private_path),
+        private_key_exists=private_path.exists(),
+        private_key_loadable=private_loadable,
+        public_key_path=str(public_path),
+        public_key_exists=public_path.exists(),
+        public_key_loadable=public_loadable,
+        auth_notify_url_configured=bool(settings.WXPAY_AUTH_NOTIFY_URL),
+        auto_reconcile_enabled=settings.WITHDRAW_AUTO_RECONCILE_ENABLED,
+        auto_reconcile_interval_sec=settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC,
+        auto_reconcile_older_than_minutes=settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES,
+        issues=issues,
+    )
+
+
+@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
+def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
+    return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
+
+
+@router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
+def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
+    order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
+    if order is None:
+        raise HTTPException(status_code=404, detail="提现单不存在")
+
+    overview = queries.get_user_overview(db, order.user_id)
+    user_snapshot = None
+    if overview is not None:
+        user = overview["user"]
+        user_snapshot = WithdrawUserSnapshot(
+            id=user.id,
+            phone=user.phone,
+            nickname=user.nickname,
+            status=user.status,
+            wechat_nickname=user.wechat_nickname,
+            wechat_avatar_url=user.wechat_avatar_url,
+            created_at=user.created_at,
+            last_login_at=user.last_login_at,
+            cash_balance_cents=overview["cash_balance_cents"],
+            withdraw_total=overview["withdraw_total"],
+            withdraw_success_cents=overview["withdraw_success_cents"],
+        )
+
+    recent_withdraws, _ = queries.list_all_withdraw_orders(
+        db, user_id=order.user_id, limit=5, cursor=None,
+    )
+    recent_cash_transactions, _ = queries.list_all_cash_transactions(
+        db, user_id=order.user_id, limit=8, cursor=None,
+    )
+    audit_logs = queries.list_withdraw_audit_logs(db, out_bill_no, limit=10)
+    risk_flags, risk_score = queries.withdraw_risk_flags(
+        order,
+        overview["user"] if overview else None,
+        recent_withdraws,
+        overview["cash_balance_cents"] if overview else 0,
+    )
+
+    return WithdrawDetailOut(
+        order=WithdrawOrderOut.model_validate(order),
+        user=user_snapshot,
+        risk_flags=risk_flags,
+        risk_score=risk_score,
+        recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
+        recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
+        audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
+    )
+
+
 # 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
 @router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
 def reconcile(
@@ -61,6 +204,159 @@ def reconcile(
     return ReconcileResult(**result)
 
 
+def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
+    success = sum(1 for item in items if item.ok)
+    return WithdrawBulkResult(
+        total=len(items),
+        success=success,
+        failed=len(items) - success,
+        items=items,
+    )
+
+
+@router.post("/bulk/refresh", response_model=WithdrawBulkResult, summary="批量刷新查单")
+def bulk_refresh_withdraws(
+    body: WithdrawBulkRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+) -> WithdrawBulkResult:
+    results: list[WithdrawBulkItemResult] = []
+    ip = get_client_ip(request)
+    for out_bill_no in body.out_bill_nos:
+        try:
+            order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
+            if order is None:
+                raise wallet_repo.WithdrawOrderNotFound
+            refreshed = wallet_repo.refresh_withdraw_status(
+                db, order.user_id, out_bill_no, cancel_if_unconfirmed=True,
+            )
+            write_audit(
+                db, admin, action="withdraw.refresh", target_type="withdraw",
+                target_id=out_bill_no,
+                detail={
+                    "status": refreshed.status,
+                    "wechat_state": refreshed.wechat_state,
+                    "fail_reason": refreshed.fail_reason,
+                    "bulk": True,
+                },
+                ip=ip, commit=True,
+            )
+            results.append(
+                WithdrawBulkItemResult(
+                    out_bill_no=out_bill_no, ok=True, status=refreshed.status
+                )
+            )
+        except wallet_repo.WithdrawOrderNotFound:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
+            )
+        except wxpay.WxPayNotConfiguredError:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="微信支付未配置")
+            )
+        except Exception as e:  # noqa: BLE001 - 批量操作单笔失败不打断整批
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
+            )
+    return _bulk_result(results)
+
+
+@router.post("/bulk/approve", response_model=WithdrawBulkResult, summary="批量审核通过并打款")
+def bulk_approve_withdraws(
+    body: WithdrawBulkRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+) -> WithdrawBulkResult:
+    results: list[WithdrawBulkItemResult] = []
+    ip = get_client_ip(request)
+    for out_bill_no in body.out_bill_nos:
+        try:
+            order = wallet_repo.approve_withdraw(db, out_bill_no)
+            write_audit(
+                db, admin, action="withdraw.approve", target_type="withdraw",
+                target_id=out_bill_no,
+                detail={
+                    "status": order.status,
+                    "wechat_state": order.wechat_state,
+                    "amount_cents": order.amount_cents,
+                    "bulk": True,
+                },
+                ip=ip, commit=True,
+            )
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
+            )
+        except wallet_repo.WithdrawOrderNotFound:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
+            )
+        except wallet_repo.WithdrawNotReviewable as e:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=str(e))
+            )
+        except wxpay.WxPayNotConfiguredError:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="微信支付未配置")
+            )
+        except Exception as e:  # noqa: BLE001 - 批量操作单笔失败不打断整批
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
+            )
+    return _bulk_result(results)
+
+
+@router.post("/bulk/reject", response_model=WithdrawBulkResult, summary="批量审核拒绝并退款")
+def bulk_reject_withdraws(
+    body: WithdrawBulkRejectRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+) -> WithdrawBulkResult:
+    results: list[WithdrawBulkItemResult] = []
+    ip = get_client_ip(request)
+    for out_bill_no in body.out_bill_nos:
+        try:
+            order = wallet_repo.reject_withdraw(db, out_bill_no, body.reason)
+            write_audit(
+                db, admin, action="withdraw.reject", target_type="withdraw",
+                target_id=out_bill_no,
+                detail={
+                    "reason": body.reason,
+                    "amount_cents": order.amount_cents,
+                    "bulk": True,
+                },
+                ip=ip, commit=True,
+            )
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
+            )
+        except wallet_repo.WithdrawOrderNotFound:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
+            )
+        except wallet_repo.WithdrawNotReviewable as e:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=str(e))
+            )
+        except Exception as e:  # noqa: BLE001 - 批量操作单笔失败不打断整批
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
+            )
+    return _bulk_result(results)
+
+
 @router.post("/{out_bill_no}/refresh", response_model=WithdrawOrderOut, summary="单笔重试查单")
 def refresh_withdraw(
     out_bill_no: str,
@@ -79,7 +375,11 @@ def refresh_withdraw(
         raise HTTPException(status_code=503, detail="微信支付未配置") from e
     write_audit(
         db, admin, action="withdraw.refresh", target_type="withdraw", target_id=out_bill_no,
-        detail={"status": refreshed.status, "wechat_state": refreshed.wechat_state},
+        detail={
+            "status": refreshed.status,
+            "wechat_state": refreshed.wechat_state,
+            "fail_reason": refreshed.fail_reason,
+        },
         ip=get_client_ip(request), commit=True,
     )
     return WithdrawOrderOut.model_validate(refreshed)

app/admin/schemas/ad_audit.py

@@ -0,0 +1,57 @@
+"""看广告金币审计 schemas。
+
+只读对账视图:把"看视频"(ad_reward_record)和"信息流"(ad_feed_reward_record)两类发奖记录,
+用与正式发奖相同的公式 [app.core.rewards.calculate_ad_reward_coin] 复算一遍 expected_coin,
+和实际入账的 actual_coin 对比,核对金币公式是否生效。字段 snake_case、金额按金币整数。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class AdCoinAuditRow(BaseModel):
+    """单条发奖记录的复算明细。"""
+
+    scene: str = Field(..., description="reward_video(看视频) / feed(信息流)")
+    record_id: int = Field(..., description="对应记录表主键")
+    user_id: int
+    created_at: datetime
+    status: str = Field(..., description="granted / capped / ecpm_missing")
+    ecpm: str | None = Field(None, description="本次采用的 eCPM 原始值(分/千次展示,SDK getEcpm 原值)")
+    ecpm_factor: float | None = Field(None, description="因子1(eCPM 档)")
+    units: int = Field(..., description="折算份数:看视频恒为 1;信息流 = 满 10 秒的份数")
+    lt_index_start: int | None = Field(None, description="本条占用的当日第几份(起)")
+    lt_index_end: int | None = Field(None, description="本条占用的当日第几份(止);看视频 = 起")
+    lt_factor_start: float | None = Field(None, description="因子2(LT)起值")
+    lt_factor_end: float | None = Field(None, description="因子2(LT)止值;看视频 = 起值")
+    expected_coin: int = Field(..., description="按公式复算应发金币")
+    actual_coin: int = Field(..., description="实际入账金币")
+    matched: bool = Field(..., description="复算与实发是否一致(capped/ecpm_missing 校验是否确为 0)")
+
+
+class AdCoinFormulaOut(BaseModel):
+    """当前金币公式参数(给前端展示规则参照)。"""
+
+    description: str = Field(
+        "eCPM元 = getEcpm分 ÷ 100;单份金币 = round(eCPM元 ÷ 1000 × 因子1 × 因子2 × coin_per_yuan);"
+        "因子1 按 eCPM元 判档(阈值 100/200/400 元)",
+        description="公式说明",
+    )
+    coin_per_yuan: int = Field(..., description="金币:元 汇率")
+    ecpm_unit: str = Field("分/千次展示(SDK getEcpm 原值)", description="eCPM 口径")
+    feed_unit_seconds: int = Field(..., description="信息流每多少秒折 1 份")
+    # [(因子值, 区间下限, 区间上限或 null)]
+    ecpm_factor_tiers: list[tuple[float, int, int | None]] = Field(..., description="因子1 档位表")
+    lt_factor_tiers: list[tuple[float, int, int | None]] = Field(..., description="因子2 LT 档位表")
+
+
+class AdCoinAuditOut(BaseModel):
+    """审计响应:公式参照 + 命中条数 + 明细。"""
+
+    date: str = Field(..., description="审计日期(北京时间 YYYY-MM-DD)")
+    formula: AdCoinFormulaOut
+    total: int = Field(..., description="返回的明细条数")
+    mismatch_count: int = Field(..., description="其中 matched=false 的条数(=0 说明公式全部生效)")
+    items: list[AdCoinAuditRow]

app/admin/schemas/dashboard.py

@@ -14,6 +14,14 @@ class DashboardUsers(BaseModel):
 
 
 class DashboardCoins(BaseModel):
+    reward_video_coin_total: int = 0
+    reward_video_watch_count: int = 0
+    feed_ad_coin_total: int = 0
+    feed_ad_watch_count: int = 0
+    signin_coin_total: int = 0
+    signin_count: int = 0
+    signin_boost_coin_total: int = 0
+    signin_boost_watch_count: int = 0
     granted_total: int
 
 

app/admin/schemas/wallet.py

@@ -5,6 +5,8 @@ from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from app.admin.schemas.admin import AdminAuditLogOut
+
 
 class CoinTxnOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
@@ -48,11 +50,99 @@ class WithdrawOrderOut(BaseModel):
     updated_at: datetime
 
 
+class WithdrawSummaryOut(BaseModel):
+    reviewing_count: int
+    reviewing_amount_cents: int
+    pending_count: int
+    failed_count: int
+    today_success_count: int
+    today_success_amount_cents: int
+    today_rejected_count: int
+
+
+class WithdrawUserSnapshot(BaseModel):
+    id: int
+    phone: str
+    nickname: str | None = None
+    status: str
+    wechat_nickname: str | None = None
+    wechat_avatar_url: str | None = None
+    created_at: datetime
+    last_login_at: datetime
+    cash_balance_cents: int
+    withdraw_total: int
+    withdraw_success_cents: int
+
+
+class WithdrawDetailOut(BaseModel):
+    order: WithdrawOrderOut
+    user: WithdrawUserSnapshot | None = None
+    risk_flags: list[str]
+    risk_score: int
+    recent_withdraws: list[WithdrawOrderOut]
+    recent_cash_transactions: list[CashTxnOut]
+    audit_logs: list[AdminAuditLogOut]
+
+
 class ReconcileResult(BaseModel):
     checked: int
     resolved: int
 
 
+class WithdrawBulkRequest(BaseModel):
+    out_bill_nos: list[str] = Field(
+        ..., min_length=1, max_length=50, description="提现商户单号列表"
+    )
+
+
+class WithdrawBulkRejectRequest(WithdrawBulkRequest):
+    reason: str = Field(
+        ..., min_length=1, max_length=200, description="批量拒绝理由(用户可见)"
+    )
+
+
+class WithdrawBulkItemResult(BaseModel):
+    out_bill_no: str
+    ok: bool
+    status: str | None = None
+    error: str | None = None
+
+
+class WithdrawBulkResult(BaseModel):
+    total: int
+    success: int
+    failed: int
+    items: list[WithdrawBulkItemResult]
+
+
+class WithdrawLedgerCheckOut(BaseModel):
+    ok: bool
+    cash_balance_total_cents: int
+    cash_transaction_total_cents: int
+    balance_diff_cents: int
+    missing_withdraw_txn_count: int
+    missing_refund_txn_count: int
+    duplicate_refund_txn_count: int
+    refund_txn_on_non_terminal_count: int
+
+
+class WxpayHealthCheckOut(BaseModel):
+    ok: bool
+    wxpay_configured: bool
+    wxpay_auth_configured: bool
+    private_key_path: str
+    private_key_exists: bool
+    private_key_loadable: bool
+    public_key_path: str
+    public_key_exists: bool
+    public_key_loadable: bool
+    auth_notify_url_configured: bool
+    auto_reconcile_enabled: bool
+    auto_reconcile_interval_sec: int
+    auto_reconcile_older_than_minutes: int
+    issues: list[str]
+
+
 class WithdrawRejectRequest(BaseModel):
     reason: str = Field(
         ..., min_length=1, max_length=200, description="拒绝理由(写入 fail_reason,用户可见)"

app/api/v1/ad.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import logging
+import json
 import uuid
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
@@ -23,6 +24,7 @@ from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories import ad_feed_reward as crud_feed
 from app.repositories import ad_reward as crud_ad
 from app.repositories import ad_watch as crud_watch
+from app.repositories import signin as crud_signin
 from app.schemas.ad import (
     AdRewardStatusOut,
     EcpmReportIn,
@@ -30,6 +32,7 @@ from app.schemas.ad import (
     FeedRewardIn,
     FeedRewardOut,
     PangleCallbackOut,
+    TestGrantIn,
     TestGrantOut,
     WatchReportIn,
     WatchReportOut,
@@ -44,6 +47,23 @@ REASON_OK = 0
 REASON_BAD_PARAMS = 1  # 验签过但缺 trans_id / user_id 非数字
 REASON_UNKNOWN_USER = 2  # user_id 不存在(可能伪造)
 
+REWARD_SCENE_REWARD_VIDEO = "reward_video"
+REWARD_SCENE_SIGNIN_BOOST = "signin_boost"
+SUPPORTED_REWARD_SCENES = {REWARD_SCENE_REWARD_VIDEO, REWARD_SCENE_SIGNIN_BOOST}
+
+
+def _parse_extra(raw_extra: str | None) -> dict[str, str]:
+    """解析客户端 setMediaExtra 透传的 JSON;旧格式/异常返回空 dict。"""
+    if not raw_extra:
+        return {}
+    try:
+        data = json.loads(raw_extra)
+    except (TypeError, ValueError):
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    return {str(k): str(v) for k, v in data.items() if v is not None}
+
 
 @router.get(
     "/pangle-callback",
@@ -52,9 +72,9 @@ REASON_UNKNOWN_USER = 2  # user_id 不存在(可能伪造)
     dependencies=[Depends(rate_limit(300, 60, "pangle-callback"))],
 )
 def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
-    """穿山甲 GroMore 在激励视频播完后回调,带 user_id / trans_id / reward_amount / sign 等 query 参数。
+    """穿山甲 GroMore 在激励视频播完后回调,带 user_id / trans_id / ecpm / sign 等 query 参数。
 
-    流程:开关/密钥就绪 → 验签(SHA256(m-key:trans_id)) → 解析 user_id/reward_amount → 幂等发金币。
+    流程:开关/密钥就绪 → 验签(SHA256(m-key:trans_id)) → 解析 user_id/场景/eCPM → 幂等发金币。
     验签失败 403(留给真请求重试);参数缺/坏或 user 不存在 → is_verify=false + reason(不发,不重试)。
     granted / capped → is_verify=true + reason=0。
     """
@@ -77,19 +97,86 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
         return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
 
     user_id = int(raw_user_id)
-    coin = rewards.resolve_ad_reward_coin(db, params.get("reward_amount"))
     raw = "&".join(f"{k}={v}" for k, v in sorted(params.items()) if k != "sign")
-    try:
-        rec = crud_ad.grant_ad_reward(
-            db, user_id, trans_id, coin=coin,
-            reward_name=params.get("reward_name"), raw=raw[:1024],
+    extra = {}
+    for extra_key in ("extra", "gromoreExtra", "gromore_extra"):
+        extra.update(_parse_extra(params.get(extra_key)))
+    reward_scene = extra.get("reward_scene") or REWARD_SCENE_REWARD_VIDEO
+    ad_session_id = extra.get("ad_session_id")
+    ecpm = params.get("ecpm")
+
+    existing = crud_ad.find_by_trans(db, trans_id)
+    if existing is not None:
+        logger.info(
+            "pangle callback idempotent user_id=%d trans_id=%s status=%s scene=%s",
+            user_id, trans_id, existing.status, existing.reward_scene,
         )
+        return PangleCallbackOut(is_verify=True, reason=REASON_OK)
+
+    try:
+        if reward_scene not in SUPPORTED_REWARD_SCENES:
+            rec = crud_ad.record_external_reward(
+                db, user_id, trans_id, coin=0, reward_scene=reward_scene[:32],
+                ad_session_id=ad_session_id, ecpm=ecpm,
+                reward_name=params.get("reward_name"), raw=raw[:1024],
+                status="unknown_scene",
+            )
+            logger.warning(
+                "pangle callback unknown scene user_id=%d trans_id=%s scene=%s",
+                user_id, trans_id, reward_scene,
+            )
+            return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
+        if reward_scene == REWARD_SCENE_SIGNIN_BOOST:
+            try:
+                boost, _balance = crud_signin.boost_today_signin(
+                    db, user_id, ad_ref_id=trans_id, commit=False
+                )
+            except crud_signin.NotSignedTodayError:
+                db.rollback()
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=0, reward_scene=reward_scene,
+                    ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    status="not_signed",
+                )
+            except crud_signin.AlreadyBoostedError:
+                db.rollback()
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=0, reward_scene=reward_scene,
+                    ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    status="already_boosted",
+                )
+            except crud_signin.LastCycleDayBoostBlockedError:
+                db.rollback()
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=0, reward_scene=reward_scene,
+                    ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    status="last_day",
+                )
+            else:
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=boost.coin_awarded,
+                    reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    commit=False,
+                )
+                db.commit()
+                db.refresh(rec)
+        else:
+            rec = crud_ad.grant_ad_reward(
+                db, user_id, trans_id, ecpm=ecpm, ad_session_id=ad_session_id,
+                reward_scene=REWARD_SCENE_REWARD_VIDEO,
+                reward_name=params.get("reward_name"), raw=raw[:1024],
+            )
     except crud_ad.UnknownUserError:
         logger.warning("pangle callback unknown user_id=%d trans_id=%s", user_id, trans_id)
         return PangleCallbackOut(is_verify=False, reason=REASON_UNKNOWN_USER)
 
     logger.info(
-        "ad reward user_id=%d trans_id=%s status=%s coin=%d", user_id, trans_id, rec.status, rec.coin
+        "ad reward user_id=%d trans_id=%s scene=%s status=%s coin=%d",
+        user_id, trans_id, rec.reward_scene, rec.status, rec.coin,
     )
     # granted / capped 均算"已处理":is_verify=true 不让穿山甲重试(capped 只是没加币)
     return PangleCallbackOut(is_verify=True, reason=REASON_OK)
@@ -153,11 +240,12 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     crud_ecpm.create_ecpm_record(
         db, user.id,
         ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
+        ad_session_id=payload.ad_session_id,
         adn=payload.adn, slot_id=payload.slot_id,
     )
     logger.info(
-        "ad ecpm report user_id=%d type=%s ecpm=%s adn=%s slot=%s",
-        user.id, payload.ad_type, payload.ecpm, payload.adn, payload.slot_id,
+        "ad ecpm report user_id=%d type=%s session=%s ecpm=%s adn=%s slot=%s",
+        user.id, payload.ad_type, payload.ad_session_id, payload.ecpm, payload.adn, payload.slot_id,
     )
     return EcpmReportOut(ok=True)
 
@@ -168,7 +256,7 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     summary="[仅本地联调]模拟穿山甲回调发奖",
     dependencies=[Depends(rate_limit(60, 60, "ad-test-grant"))],
 )
-def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
+def test_grant(user: CurrentUser, db: DbSession, payload: TestGrantIn | None = None) -> TestGrantOut:
     """⚠️ 仅本地联调用:没部署公网、穿山甲 S2S 回调打不到本地时,客户端(debug 包)看完广告后
     调这个接口,直接走与回调相同的发奖逻辑(幂等 + 每日上限),验证"看广告→金币到账"全链路。
 
@@ -178,18 +266,66 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
     if not settings.AD_REWARD_TEST_GRANT_ENABLED:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="not found")
 
-    # 每次新 trans_id,模拟一次独立的穿山甲发奖回调(幂等键各不相同 → 每次都发,直到当日上限)
+    reward_scene = (payload.reward_scene if payload is not None else REWARD_SCENE_REWARD_VIDEO)
+    if reward_scene not in SUPPORTED_REWARD_SCENES:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="bad reward_scene")
+
+    # 每次新 trans_id,模拟一次独立的穿山甲发奖回调(幂等键各不相同 → 每次都发,直到当日上限/今日膨胀一次)
     trans_id = f"test-{user.id}-{uuid.uuid4().hex}"
-    try:
-        rec = crud_ad.grant_ad_reward(
-            db, user.id, trans_id, reward_name="测试发奖", raw="client debug test-grant"
-        )
-    except crud_ad.UnknownUserError as e:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
+    if reward_scene == REWARD_SCENE_SIGNIN_BOOST:
+        try:
+            boost, _balance = crud_signin.boost_today_signin(
+                db, user.id, ad_ref_id=trans_id, commit=False
+            )
+        except crud_signin.NotSignedTodayError:
+            db.rollback()
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=0, reward_scene=reward_scene,
+                raw="client debug test-grant signin_boost", status="not_signed",
+            )
+        except crud_signin.AlreadyBoostedError:
+            db.rollback()
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=0, reward_scene=reward_scene,
+                raw="client debug test-grant signin_boost", status="already_boosted",
+            )
+        except crud_signin.LastCycleDayBoostBlockedError:
+            db.rollback()
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=0, reward_scene=reward_scene,
+                raw="client debug test-grant signin_boost", status="last_day",
+            )
+        else:
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=boost.coin_awarded,
+                reward_scene=reward_scene, reward_name="测试签到膨胀",
+                raw="client debug test-grant signin_boost", commit=False,
+            )
+            db.commit()
+            db.refresh(rec)
+    else:
+        # 优先用客户端按 ad_session_id 上报的真实 eCPM(走与正式发奖相同的公式);
+        # 取不到或 eCPM≤0(测试应用常返 0/假值)时兜底 200,保证本地联调仍能验出非零金币。
+        ad_session_id = payload.ad_session_id if payload is not None else None
+        ecpm_val = "200"
+        if ad_session_id:
+            ecpm_rec = crud_ecpm.find_by_session(db, user_id=user.id, ad_session_id=ad_session_id)
+            if ecpm_rec is not None and rewards.parse_ecpm_fen(ecpm_rec.ecpm_raw) > 0:
+                ecpm_val = ecpm_rec.ecpm_raw
+        try:
+            rec = crud_ad.grant_ad_reward(
+                db, user.id, trans_id, ecpm=ecpm_val, ad_session_id=ad_session_id,
+                reward_name="测试发奖", raw=f"client debug test-grant ecpm={ecpm_val}",
+            )
+        except crud_ad.UnknownUserError as e:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
 
     (used, limit, coin_per, round_count, cooldown_until,
      _watched, _watch_limit) = crud_ad.today_status(db, user.id)
-    logger.info("ad TEST grant user_id=%d status=%s coin=%d", user.id, rec.status, rec.coin)
+    logger.info(
+        "ad TEST grant user_id=%d scene=%s status=%s coin=%d",
+        user.id, reward_scene, rec.status, rec.coin,
+    )
     return TestGrantOut(
         granted=(rec.status == "granted"),
         status=rec.status,
@@ -220,6 +356,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         client_event_id=payload.client_event_id,
         ecpm=payload.ecpm,
         duration_seconds=payload.duration_seconds,
+        ad_session_id=payload.ad_session_id,
         adn=payload.adn,
         slot_id=payload.slot_id,
     )

app/api/v1/coupon.py

@@ -183,7 +183,15 @@ def coupon_prompt_should_show(
     device_id: str, db: DbSession
 ) -> CouponPromptShouldShowOut:
     """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
-    (前台 SP 缓存做快速路径,这里是权威)。"""
+    (纯后台判据,客户端不再做前台 SP 缓存判断)。"""
     return CouponPromptShouldShowOut(
         should_show=not coupon_repo.has_engaged_today(db, device_id)
     )
+
+
+@router.post("/prompt/reset", summary="重置今日领券引导窗 engagement(开发测频控用)")
+def coupon_prompt_reset(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
+    """删这台设备今天的 engagement → has_engaged_today 变 false,今天又能弹。
+    开发设置「重置今日领券弹窗状态」按钮调。MVP 不鉴权,按 device_id。"""
+    coupon_repo.reset_today_engagement(db, payload.device_id)
+    return {"ok": True}

app/api/v1/signin.py

@@ -12,7 +12,9 @@ import logging
 from fastapi import APIRouter, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.repositories import ad_reward as crud_ad
 from app.repositories import signin as crud_signin
+from app.repositories import wallet as crud_wallet
 from app.schemas.welfare import (
     SigninBoostRequest,
     SigninBoostResultOut,
@@ -54,14 +56,21 @@ def do_signin(user: CurrentUser, db: DbSession) -> SigninResultOut:
 def boost_signin(
     payload: SigninBoostRequest, user: CurrentUser, db: DbSession
 ) -> SigninBoostResultOut:
-    try:
-        record, balance = crud_signin.boost_today_signin(
-            db, user.id, ad_ref_id=payload.ad_ref_id
-        )
-    except crud_signin.NotSignedTodayError as e:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="not signed today") from e
-    except crud_signin.AlreadyBoostedError as e:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="already boosted today") from e
+    if not payload.ad_ref_id:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="ad reward required")
+    ad_rec = crud_ad.find_by_trans(db, payload.ad_ref_id)
+    if (
+        ad_rec is None
+        or ad_rec.user_id != user.id
+        or ad_rec.reward_scene != "signin_boost"
+        or ad_rec.status != "granted"
+    ):
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="ad reward not verified")
+    record = crud_signin.boost_by_ad_ref(db, user.id, payload.ad_ref_id)
+    if record is None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="signin boost not granted")
+    acc = crud_wallet.get_or_create_account(db, user.id)
+    balance = acc.coin_balance
 
     logger.info(
         "signin boost ok user_id=%d date=%s coin=%d",

app/api/v1/wallet.py

@@ -176,7 +176,7 @@ def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
     "/withdraw",
     response_model=WithdrawResultOut,
     summary="发起提现(扣款建单,待人工审核;审核通过后才打款)",
-    dependencies=[Depends(rate_limit(20, 60, "withdraw"))],  # #6 同 IP 每分钟≤20 次
+    dependencies=[Depends(rate_limit(5, 60, "withdraw"))],  # IP 级粗限流;用户级未完成单限制在仓库层
 )
 def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut:
     # 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦
@@ -189,12 +189,17 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
     except crud_wallet.InvalidWithdrawAmountError as e:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"amount_cents must be within [{WITHDRAW_MIN_CENTS}, {WITHDRAW_MAX_CENTS}]",
+            detail="提现金额不符合要求",
         ) from e
     except crud_wallet.WechatNotBoundError as e:
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先绑定微信") from e
+    except crud_wallet.WithdrawTooFrequentError as e:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="已有提现申请正在审核或打款中,请处理完成后再申请",
+        ) from e
     except crud_wallet.InsufficientCashError as e:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="insufficient cash balance") from e
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="现金余额不足") from e
 
     # 调试直发(非生产):skip_review=true 时跳过人工审核,立即发起微信转账(等价 admin approve)。
     # 双闸保护——客户端仅 debug 包下发此 flag,服务端仅非 prod 才认;任一道闸拦住即恢复正常审核,

app/core/config.py

@@ -102,6 +102,9 @@ class Settings(BaseSettings):
     WXPAY_PUBLIC_KEY_PATH: str = "./secrets/pub_key.pem"             # 微信支付平台公钥
     WXPAY_TRANSFER_SCENE_ID: str = "1000"  # 转账场景 ID(1000=现金营销)
     WXPAY_REQUEST_TIMEOUT_SEC: int = 10
+    WITHDRAW_AUTO_RECONCILE_ENABLED: bool = False
+    WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC: int = 300
+    WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES: int = 15
     # 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
     # 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
     # (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。

app/core/config_schema.py

@@ -42,7 +42,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     "ad_reward_coin": {
         "default": r.AD_REWARD_COIN, "label": "看广告单次金币",
         "group": "看广告", "type": "int",
-        "help": "看完一个激励视频发的金币;须与穿山甲后台代码位'奖励数量'保持一致。",
+        "help": "历史兼容/测试展示值;正式发放按 eCPM 公式计算。",
     },
     "ad_daily_limit": {
         "default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
@@ -50,7 +50,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     },
     "ad_max_coin": {
         "default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
-        "group": "看广告", "type": "int", "help": "夹紧穿山甲回调里异常的 reward_amount,防刷爆。",
+        "group": "看广告", "type": "int", "help": "历史 reward_amount 口径保留项;正式发放按 eCPM 公式计算。",
     },
     "ad_round_count": {
         "default": r.VIDEO_ROUND_REQUIRED_COUNT, "label": "每轮看广告次数",
@@ -60,4 +60,9 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
         "default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "广告关闭后冷却(秒)",
         "group": "看广告", "type": "int", "help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
     },
+    "signin_boost_coin": {
+        "default": r.SIGNIN_BOOST_COIN, "label": "签到膨胀固定金币",
+        "group": "签到", "type": "int",
+        "help": "Day1-Day13 签到后看完激励视频额外发放的固定金币;Day14 不展示也不允许膨胀。",
+    },
 }

app/core/rewards.py

@@ -92,7 +92,11 @@ INVITE_FP_WINDOW_DAYS: int = 7
 
 
 # ===== 看激励视频 / 信息流广告发金币 =====
-# 金币数值体系约定:eCPM 单位按"元/千次展示"处理,单次收入 = eCPM / 1000 元。
+# eCPM 取自穿山甲 SDK getShowEcpm().getEcpm(),官方口径单位是【分/千次展示】(不是元!
+# csjplatform 文档原文"通过 getEcpm 获取的单位是分")。计算时先 ÷100 转成元;
+# 因子1 档位阈值按【元/千次】定(100/200/400 元 = ¥100/¥200/¥400 CPM,产品口径 2026-06-09)。
+# 注:真实 eCPM 一般 <¥100 CPM,故多落最低档 0.1,高档基本不触发——这是产品有意的取舍。
+# 单次展示收益(元) = eCPM元 ÷ 1000(每千次→单次)。
 AD_ECPM_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
     (0.1, 0, 100),
     (0.3, 101, 200),
@@ -108,8 +112,8 @@ AD_LT_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
 )
 
 
-def parse_ecpm_yuan(ecpm: str | int | float | None) -> float:
-    """解析 eCPM 原始值。当前产品口径:SDK 返回值按"元/千次展示"处理。"""
+def parse_ecpm_fen(ecpm: str | int | float | None) -> float:
+    """解析 eCPM 原始值(穿山甲 getEcpm 原值,单位=分/千次展示)。非法/缺失→0。"""
     if ecpm is None:
         return 0.0
     try:
@@ -119,8 +123,14 @@ def parse_ecpm_yuan(ecpm: str | int | float | None) -> float:
     return max(0.0, value)
 
 
+def parse_ecpm_yuan(ecpm: str | int | float | None) -> float:
+    """eCPM 转成元(getEcpm 原值是分,÷100)。因子档位判定与收益换算都用元。"""
+    return parse_ecpm_fen(ecpm) / 100.0
+
+
 def ad_ecpm_factor(ecpm_yuan: float) -> float:
-    """eCPM 档位因子:0-100=0.1,101-200=0.3,201-400=0.4,>400=0.6。"""
+    """eCPM 档位因子(阈值单位=元/千次):≤100=0.1,101-200=0.3,201-400=0.4,>400=0.6。
+    产品口径(2026-06-09):阈值按元判档;真实 eCPM(<¥100 CPM)多落最低档 0.1。"""
     if ecpm_yuan > 400:
         return 0.6
     if ecpm_yuan > 200:
@@ -142,21 +152,23 @@ def ad_lt_factor(today_count_after_this: int) -> float:
 def calculate_ad_reward_coin(ecpm: str | int | float | None, today_count_after_this: int) -> int:
     """按金币数值体系计算单份广告奖励金币。
 
-    单次奖励(元)=eCPM/1000 × 因子1(eCPM 档) × 因子2(LT);再按 1 元=10000 金币取整。
+    eCPM 是穿山甲 getEcpm 原值,单位【分/千次展示】;先 ÷100 转成元(因子判档 + 收益换算都用元)。
+    单次收益(元)= eCPM元 ÷ 1000(每千次→单次) × 因子1(eCPM 元档) × 因子2(LT);
+    再按 1 元=10000 金币取整。
     """
     ecpm_yuan = parse_ecpm_yuan(ecpm)
     yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(today_count_after_this)
     return max(0, round(yuan * COIN_PER_YUAN))
 
 
+SIGNIN_BOOST_COIN: int = 2000
+
+
 # ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
-# 看完一个激励视频发的金币(666 金币 ≈¥0.0666,汇率 10000 金币=1 元)。
-# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
-#       ③ test-grant 本地联调的发奖额。
-# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
-# 代码位"奖励数量"配成与本值一致(=666),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
+# 历史固定金币口径保留项。正式激励视频实发按 calculate_ad_reward_coin(eCPM, 当日第 N 次)
+# 计算;该值只用于旧接口兼容、配置页展示和本地联调兜底。
 AD_REWARD_COIN: int = 666
-# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
+# 历史 reward_amount 口径保留项。正式激励视频实发按 eCPM 公式计算。
 MAX_AD_REWARD_COIN: int = 1000
 # 每用户每日发奖次数上限。产品口径:一天最多看 500 次广告。
 DAILY_AD_REWARD_LIMIT: int = 500
@@ -174,7 +186,7 @@ VIDEO_ROUND_COOLDOWN_SECONDS: int = 3
 
 
 def resolve_ad_reward_coin(db, reward_amount: str | int | None) -> int:  # noqa: ANN001
-    """把穿山甲回调的 reward_amount(后台代码位配的"奖励数量")解析成本次发放金币。
+    """历史 reward_amount 口径解析函数,保留给旧脚本/旧配置兼容。
 
     缺失 / 非数字 / ≤0 → 回退配置单次金币;超过配置单次上限 → 夹紧(均从 app_config 读)。
     注:reward_amount 不参与穿山甲 sign(只签 trans_id),但伪造回调需先伪造合法 sign
@@ -241,3 +253,7 @@ def get_ad_round_count(db) -> int:  # noqa: ANN001
 
 def get_ad_cooldown_sec(db) -> int:  # noqa: ANN001
     return int(_cfg(db, "ad_cooldown_sec"))
+
+
+def get_signin_boost_coin(db) -> int:  # noqa: ANN001
+    return int(_cfg(db, "signin_boost_coin"))

app/core/withdraw_reconcile_worker.py

@@ -0,0 +1,118 @@
+"""提现 pending 单自动对账后台任务。"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.integrations.wxpay import WxPayNotConfiguredError
+from app.repositories import wallet as wallet_repo
+
+logger = logging.getLogger("shagua.withdraw_reconcile")
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "withdraw_reconcile.lock"
+
+
+def _touch_lock() -> None:
+    with contextlib.suppress(FileNotFoundError):
+        os.utime(_LOCK_PATH, None)
+
+
+@contextlib.contextmanager
+def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
+    """同机多进程保护:同一时间只允许一个自动查单 worker 运行。"""
+    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd: int | None = None
+    try:
+        try:
+            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        except FileExistsError:
+            try:
+                age = time.time() - _LOCK_PATH.stat().st_mtime
+            except FileNotFoundError:
+                age = stale_after_sec + 1
+            if age > stale_after_sec:
+                with contextlib.suppress(FileNotFoundError):
+                    _LOCK_PATH.unlink()
+                try:
+                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                except FileExistsError:
+                    fd = None
+
+        if fd is None:
+            yield False
+            return
+
+        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
+        yield True
+    finally:
+        if fd is not None:
+            os.close(fd)
+            with contextlib.suppress(FileNotFoundError):
+                _LOCK_PATH.unlink()
+
+
+def _reconcile_once(older_than_minutes: int) -> dict:
+    with SessionLocal() as db:
+        return wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
+
+
+async def _run_loop() -> None:
+    interval = max(30, int(settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC))
+    older_than = max(1, int(settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES))
+    lock_stale_after = max(interval * 3, 600)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning("withdraw auto reconcile skipped: another worker owns lock")
+            return
+
+        await _run_locked_loop(interval, older_than)
+
+
+async def _run_locked_loop(interval: int, older_than: int) -> None:
+    logger.info(
+        "withdraw auto reconcile started interval=%ss older_than=%sm",
+        interval,
+        older_than,
+    )
+    try:
+        while True:
+            try:
+                _touch_lock()
+                result = await asyncio.to_thread(_reconcile_once, older_than)
+                if result["checked"] or result["resolved"]:
+                    logger.info("withdraw auto reconcile result=%s", result)
+            except WxPayNotConfiguredError:
+                logger.warning("withdraw auto reconcile skipped: wxpay not configured")
+            except SQLAlchemyError:
+                logger.exception("withdraw auto reconcile db error")
+            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
+                logger.exception("withdraw auto reconcile unexpected error")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("withdraw auto reconcile stopped")
+        raise
+
+
+def start_withdraw_reconcile_worker() -> asyncio.Task | None:
+    if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
+        logger.info("withdraw auto reconcile disabled")
+        return None
+    if not settings.wxpay_configured:
+        logger.warning("withdraw auto reconcile enabled but wxpay not configured")
+    return asyncio.create_task(_run_loop(), name="withdraw-auto-reconcile")
+
+
+async def stop_withdraw_reconcile_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/integrations/pangle.py

@@ -8,9 +8,9 @@
 我们客户端是 useMediation(true) 融合,回调走 GroMore 广告位层级,**不是**联盟代码位层级):
 **sign = SHA256("{m-key}:{trans_id}")** 的十六进制串。注意:
 - 是**普通 SHA256**,不是 HMAC,也不是 RSA;
-- **只对 `m-key:trans_id` 这一个字符串**签名,user_id / reward_amount 等其余参数**不参与**签名
+- **只对 `m-key:trans_id` 这一个字符串**签名,user_id / ecpm / extra 等其余参数**不参与**签名
   (它们的可信度由"能算出正确 sign = 知道 m-key"间接保证——伪造者没有 m-key 就连合法 sign
-  都造不出,自然也无法注入假 user_id/reward_amount)。
+  都造不出,自然也无法注入假 user_id/ecpm/extra)。
 m-key(安全密钥)在穿山甲后台「GroMore 聚合管理 → 搜广告位ID → 编辑」处获取,配到
 `settings.PANGLE_REWARD_SECRET`。(联盟代码位层级用的是另一套 Security Key + isValid 响应,我们不走那条。)
 """

app/integrations/wxpay.py

@@ -14,7 +14,9 @@ import base64
 import json
 import time
 import uuid
+from pathlib import Path
 
+import certifi
 import httpx
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding
@@ -23,6 +25,7 @@ from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPubl
 from app.core.config import settings
 
 _API_HOST = "https://api.mch.weixin.qq.com"
+_PROJECT_ROOT = Path(__file__).resolve().parents[2]
 _TRANSFER_PATH = "/v3/fund-app/mch-transfer/transfer-bills"
 # 免确认收款授权(用户授权免确认模式)
 _AUTH_PATH = "/v3/fund-app/mch-transfer/user-confirm-authorization"
@@ -38,15 +41,30 @@ class WxPayNotConfiguredError(Exception):
     """微信支付凭证 / 证书缺失,无法调用。"""
 
 
+def _http_client() -> httpx.Client:
+    """微信相关 HTTP 客户端。
+
+    显式使用当前 Python 环境的 certifi 证书,避免被外部 SSL_CERT_FILE 指到不存在文件。
+    代理仍允许从环境变量读取,方便本地开发。
+    """
+    return httpx.Client(verify=certifi.where())
+
+
+def _resolve_config_path(value: str) -> Path:
+    path = Path(value)
+    return path if path.is_absolute() else _PROJECT_ROOT / path
+
+
 def _load_private_key() -> RSAPrivateKey:
     global _private_key
     if _private_key is None:
+        key_path = _resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH)
         try:
-            with open(settings.WXPAY_MCH_PRIVATE_KEY_PATH, "rb") as f:
+            with key_path.open("rb") as f:
                 _private_key = serialization.load_pem_private_key(f.read(), password=None)
         except FileNotFoundError as e:
             raise WxPayNotConfiguredError(
-                f"商户私钥不存在: {settings.WXPAY_MCH_PRIVATE_KEY_PATH}"
+                f"商户私钥不存在: {key_path}"
             ) from e
     return _private_key
 
@@ -54,12 +72,13 @@ def _load_private_key() -> RSAPrivateKey:
 def _load_public_key() -> RSAPublicKey:
     global _public_key
     if _public_key is None:
+        key_path = _resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH)
         try:
-            with open(settings.WXPAY_PUBLIC_KEY_PATH, "rb") as f:
+            with key_path.open("rb") as f:
                 _public_key = serialization.load_pem_public_key(f.read())
         except FileNotFoundError as e:
             raise WxPayNotConfiguredError(
-                f"微信支付平台公钥不存在: {settings.WXPAY_PUBLIC_KEY_PATH}"
+                f"微信支付平台公钥不存在: {key_path}"
             ) from e
     return _public_key
 
@@ -126,7 +145,7 @@ def create_transfer(
         "Content-Type": "application/json",
         "Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.post(
             f"{_API_HOST}{_TRANSFER_PATH}",
             content=body_str,
@@ -143,7 +162,7 @@ def query_transfer(out_bill_no: str) -> dict:
         "Authorization": _build_authorization("GET", path, ""),
         "Accept": "application/json",
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.get(
             f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
         )
@@ -159,7 +178,7 @@ def cancel_transfer(out_bill_no: str) -> dict:
         "Accept": "application/json",
         "Content-Type": "application/json",
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.post(
             f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
         )
@@ -171,16 +190,17 @@ def code_to_userinfo(code: str) -> dict:
     返回 {openid, nickname, avatar_url, raw}。失败抛 ValueError。
     注意:微信隐私新政下,部分 app 的 sns/userinfo 可能返回脱敏值(昵称"微信用户"/灰头像);
     nickname/avatar_url 可能为空,调用方需兜底。"""
-    r1 = httpx.get(
-        "https://api.weixin.qq.com/sns/oauth2/access_token",
-        params={
-            "appid": settings.WECHAT_APP_ID,
-            "secret": settings.WECHAT_APP_SECRET,
-            "code": code,
-            "grant_type": "authorization_code",
-        },
-        timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
-    )
+    with _http_client() as client:
+        r1 = client.get(
+            "https://api.weixin.qq.com/sns/oauth2/access_token",
+            params={
+                "appid": settings.WECHAT_APP_ID,
+                "secret": settings.WECHAT_APP_SECRET,
+                "code": code,
+                "grant_type": "authorization_code",
+            },
+            timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
+        )
     d1 = r1.json()
     if "openid" not in d1 or "access_token" not in d1:
         raise ValueError(f"微信授权失败: {d1.get('errmsg', d1)}")
@@ -191,11 +211,12 @@ def code_to_userinfo(code: str) -> dict:
     raw: dict = {}
     # userinfo 拉取失败不应让绑定失败(openid 已拿到),吞掉异常只是没昵称头像
     try:
-        r2 = httpx.get(
-            "https://api.weixin.qq.com/sns/userinfo",
-            params={"access_token": d1["access_token"], "openid": openid, "lang": "zh_CN"},
-            timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
-        )
+        with _http_client() as client:
+            r2 = client.get(
+                "https://api.weixin.qq.com/sns/userinfo",
+                params={"access_token": d1["access_token"], "openid": openid, "lang": "zh_CN"},
+                timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
+            )
         raw = r2.json()
         nickname = raw.get("nickname") or None
         avatar_url = raw.get("headimgurl") or None
@@ -244,7 +265,7 @@ def apply_transfer_authorization(
         "Accept": "application/json",
         "Content-Type": "application/json",
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.post(
             f"{_API_HOST}{_AUTH_PATH}",
             content=body_str,
@@ -262,7 +283,7 @@ def query_transfer_authorization(out_authorization_no: str) -> dict:
         "Authorization": _build_authorization("GET", path, ""),
         "Accept": "application/json",
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.get(
             f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
         )
@@ -277,7 +298,7 @@ def close_transfer_authorization(out_authorization_no: str) -> dict:
         "Accept": "application/json",
         "Content-Type": "application/json",
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.post(
             f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC
         )
@@ -319,7 +340,7 @@ def pre_transfer_with_authorization(
         "Content-Type": "application/json",
         "Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.post(
             f"{_API_HOST}{_PRE_TRANSFER_AUTH_PATH}",
             content=body_str,
@@ -356,7 +377,7 @@ def transfer_with_authorization(
         "Content-Type": "application/json",
         "Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID,
     }
-    with httpx.Client() as client:
+    with _http_client() as client:
         resp = client.post(
             f"{_API_HOST}{_TRANSFER_WITH_AUTH_PATH}",
             content=body_str,

app/main.py

@@ -35,6 +35,10 @@ from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
 from app.core.config import settings
 from app.core.logging import setup_logging
+from app.core.withdraw_reconcile_worker import (
+    start_withdraw_reconcile_worker,
+    stop_withdraw_reconcile_worker,
+)
 
 setup_logging(debug=settings.APP_DEBUG)
 logger = logging.getLogger("shagua.main")
@@ -50,8 +54,12 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.APP_DEBUG,
         settings.DATABASE_URL.split("://", 1)[0],
     )
-    yield
-    logger.info("shutting down")
+    reconcile_task = start_withdraw_reconcile_worker()
+    try:
+        yield
+    finally:
+        await stop_withdraw_reconcile_worker(reconcile_task)
+        logger.info("shutting down")
 
 
 app = FastAPI(

app/models/ad_ecpm.py

@@ -1,11 +1,8 @@
 """广告展示 eCPM 上报记录(内部收益统计/对账)。
 
-每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM 信息。和发奖记录
-[ad_reward.AdRewardRecord] 是**两条独立的数据流**:
-- 发奖走穿山甲 S2S 回调(后端 → 有 trans_id、无 ecpm);
-- eCPM 走客户端上报(客户端 → 有 ecpm、无 trans_id)。
-两者没有公共键,无法逐条一一对应,所以本表用于**按用户/按天聚合收益**口径的对账,
-不做"这条发奖 = 这条 ecpm"的精确关联。穿山甲后台报表才是结算权威,本表是细粒度补充。
+每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM 信息。`ad_session_id`
+由客户端生成,并通过激励视频 extra 透传给 S2S 回调,用于把"展示 eCPM"与
+"奖励完成"绑定。没有 session id 的旧上报仍可作为按天对账补充。
 
 ⚠️ `ecpm_raw` 原样存客户端上报的字符串——eCPM 单位(分 / 元)截至 2026-05-31 尚未最终确认,
 确认后再加一列解析好的数值;在此之前对账按"待定单位"处理。
@@ -14,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import DateTime, ForeignKey, Integer, String, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -22,6 +19,9 @@ from app.db.base import Base
 
 class AdEcpmRecord(Base):
     __tablename__ = "ad_ecpm_record"
+    __table_args__ = (
+        UniqueConstraint("ad_session_id", name="uq_ad_ecpm_record_session"),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     user_id: Mapped[int] = mapped_column(
@@ -29,11 +29,13 @@ class AdEcpmRecord(Base):
     )
     # 广告类型:reward_video(激励视频) / draw(Draw 信息流) 等;不强行统一代码位,各类型各自上报
     ad_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    # 客户端生成的一次广告会话 id;激励视频 S2S 回调 extra 会透传同值
+    ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     # 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)
     adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
     # 实际展示用的代码位(底层 mediation rit,非客户端配置位)
     slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    # 客户端上报的 eCPM 原始字符串(单位待确认,原样存)
+    # 客户端上报的 eCPM 原始字符串(单位:分/千次展示,SDK getEcpm 原值,原样存)
     ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
     # 北京时间日期串 'YYYY-MM-DD',按它等值做"按天聚合"(不在 SQL 里做跨时区 date 比较)
     report_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)

app/models/ad_feed_reward.py

@@ -20,6 +20,7 @@ class AdFeedRewardRecord(Base):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     client_event_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     user_id: Mapped[int] = mapped_column(
         Integer, ForeignKey("user.id"), index=True, nullable=False
     )

app/models/ad_reward.py

@@ -1,8 +1,8 @@
 """看激励视频发奖记录(穿山甲 S2S 回调)。
 
-每条 = 穿山甲一次发奖回调。`trans_id`(穿山甲交易号)唯一,做幂等键:穿山甲会重试回调,
-同号只发一次金币。`reward_date`(北京时间日期串)给"每日上限"计数用——按日期串等值查,
-不在 SQL 里做跨时区 date 比较(SQLite 上不可靠)。审计 / 对账时整张表可逐笔回溯。
+每条 = 穿山甲/GroMore 一次服务端激励回调。`trans_id`(交易号)唯一,做幂等键:
+穿山甲会重试回调,同号只处理一次。`reward_scene` 区分福利页激励视频、签到膨胀等
+不同奖励场景,避免统计和每日上限互相污染。
 """
 from __future__ import annotations
 
@@ -25,8 +25,14 @@ class AdRewardRecord(Base):
     )
     # 实发金币(capped 时为 0)
     coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # granted(已发) / capped(当日超限未发)
+    # granted(已发) / capped(当日超限未发) / ecpm_missing(缺 eCPM 未发)
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="granted")
+    # reward_video(福利页看视频) / signin_boost(签到膨胀)
+    reward_scene: Mapped[str] = mapped_column(String(32), nullable=False, default="reward_video")
+    # 客户端生成并通过 extra 透传的广告会话 id
+    ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+    # 本次发奖采用的 eCPM 原始值(回调自带或按 ad_session_id 匹配的客户端上报)
+    ecpm_raw: Mapped[str | None] = mapped_column(String(32), nullable=True)
     # 北京时间日期串 'YYYY-MM-DD',按它等值统计当日发奖次数
     reward_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
     # 穿山甲上报的奖励名(参考,不作发奖依据)
@@ -39,4 +45,7 @@ class AdRewardRecord(Base):
     )
 
     def __repr__(self) -> str:  # pragma: no cover
-        return f"<AdRewardRecord trans_id={self.trans_id} user_id={self.user_id} {self.status} coin={self.coin}>"
+        return (
+            f"<AdRewardRecord trans_id={self.trans_id} user_id={self.user_id} "
+            f"scene={self.reward_scene} {self.status} coin={self.coin}>"
+        )

app/models/wallet.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, func, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -76,6 +76,15 @@ class WithdrawOrder(Base):
     """
 
     __tablename__ = "withdraw_order"
+    __table_args__ = (
+        Index(
+            "ux_withdraw_order_user_active",
+            "user_id",
+            unique=True,
+            sqlite_where=text("status IN ('reviewing', 'pending')"),
+            postgresql_where=text("status IN ('reviewing', 'pending')"),
+        ),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     user_id: Mapped[int] = mapped_column(
@@ -148,6 +157,15 @@ class CashTransaction(Base):
     """现金流水(单位:分)。金币兑现金、提现都记在这里。"""
 
     __tablename__ = "cash_transaction"
+    __table_args__ = (
+        Index(
+            "ux_cash_transaction_withdraw_refund_ref",
+            "ref_id",
+            unique=True,
+            sqlite_where=text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"),
+            postgresql_where=text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"),
+        ),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     user_id: Mapped[int] = mapped_column(

app/repositories/ad_ecpm.py

@@ -7,6 +7,7 @@ user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条
 from __future__ import annotations
 
 from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.core.rewards import cn_today
@@ -19,24 +20,52 @@ def create_ecpm_record(
     *,
     ad_type: str,
     ecpm_raw: str,
+    ad_session_id: str | None = None,
     adn: str | None = None,
     slot_id: str | None = None,
 ) -> AdEcpmRecord:
     """落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。"""
+    if ad_session_id:
+        existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+        if existing is not None:
+            return existing
     rec = AdEcpmRecord(
         user_id=user_id,
         ad_type=ad_type,
+        ad_session_id=ad_session_id,
         adn=adn,
         slot_id=slot_id,
         ecpm_raw=ecpm_raw,
         report_date=cn_today().isoformat(),
     )
     db.add(rec)
-    db.commit()
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        if ad_session_id:
+            existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+            if existing is not None:
+                return existing
+        raise
     db.refresh(rec)
     return rec
 
 
+def find_by_session(
+    db: Session, *, user_id: int, ad_session_id: str | None
+) -> AdEcpmRecord | None:
+    """按广告会话找 eCPM。旧客户端无 ad_session_id 时返回 None。"""
+    if not ad_session_id:
+        return None
+    return db.execute(
+        select(AdEcpmRecord).where(
+            AdEcpmRecord.user_id == user_id,
+            AdEcpmRecord.ad_session_id == ad_session_id,
+        )
+    ).scalar_one_or_none()
+
+
 def count_today(db: Session, user_id: int) -> int:
     """该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
     return db.execute(

app/repositories/ad_feed_reward.py

@@ -63,6 +63,7 @@ def grant_feed_reward(
     client_event_id: str,
     ecpm: str,
     duration_seconds: int,
+    ad_session_id: str | None = None,
     adn: str | None = None,
     slot_id: str | None = None,
 ) -> AdFeedRewardRecord:
@@ -82,6 +83,7 @@ def grant_feed_reward(
             reward_date=today,
             duration_seconds=safe_duration,
             unit_count=unit_count,
+            ad_session_id=ad_session_id,
             ecpm_raw=ecpm,
             adn=adn,
             slot_id=slot_id,
@@ -103,6 +105,7 @@ def grant_feed_reward(
         reward_date=today,
         duration_seconds=safe_duration,
         unit_count=unit_count,
+        ad_session_id=ad_session_id,
         ecpm_raw=ecpm,
         adn=adn,
         slot_id=slot_id,

app/repositories/ad_reward.py

@@ -5,8 +5,8 @@
   2. trans_id 唯一 → 同一交易号二次回调不重复发(穿山甲会重试)。
   3. 当日发奖次数 ≥ 上限 → 记一条 status='capped' 但不发金币。
 
-发金币复用 `wallet.grant_coins`(biz_type='ad_reward', ref_id=trans_id),与发奖记录同事务,
-保证"记一笔 + 加金币"原子化。单次金币 / 每日上限 / 每轮冷却 都从 app_config 读(运营后台可改)。
+普通福利页激励视频按 eCPM 公式发金币;缺 eCPM 时记 status='ecpm_missing' 但不发金币。
+签到膨胀等其他激励视频场景复用本表记录 S2S 幂等,实际发币由各自业务仓储完成。
 """
 from __future__ import annotations
 
@@ -21,6 +21,7 @@ from app.core.ad_cooldown import compute_cooldown
 from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
+from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories import wallet as crud_wallet
 from app.repositories.ad_watch import watched_seconds_today
 
@@ -35,6 +36,11 @@ def _find_by_trans(db: Session, trans_id: str) -> AdRewardRecord | None:
     ).scalar_one_or_none()
 
 
+def find_by_trans(db: Session, trans_id: str) -> AdRewardRecord | None:
+    """按 S2S 交易号查询处理记录,供路由在分场景前做幂等短路。"""
+    return _find_by_trans(db, trans_id)
+
+
 def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
     return db.execute(
         select(func.count())
@@ -43,6 +49,7 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
             AdRewardRecord.user_id == user_id,
             AdRewardRecord.reward_date == reward_date,
             AdRewardRecord.status == "granted",
+            AdRewardRecord.reward_scene == "reward_video",
         )
     ).scalar_one()
 
@@ -52,16 +59,13 @@ def grant_ad_reward(
     user_id: int,
     trans_id: str,
     *,
-    coin: int | None = None,
+    ecpm: str | None = None,
+    ad_session_id: str | None = None,
+    reward_scene: str = "reward_video",
     reward_name: str | None = None,
     raw: str | None = None,
 ) -> AdRewardRecord:
-    """看广告发奖(幂等 + 每日限额)。返回发奖记录(status=granted/capped)。
-
-    coin 为本次发放金币(由调用方按穿山甲回调 reward_amount 解析,见
-    rewards.resolve_ad_reward_coin);None → 读配置 get_ad_reward_coin(test-grant / 缺省场景)。
-    user_id 不存在抛 UnknownUserError(不建账户,防伪造 user_id 刷出脏数据)。
-    """
+    """福利页激励视频发奖(幂等 + 每日限额 + eCPM 公式)。"""
     # #2 幂等:同 trans_id 已处理过 → 原样返回,不重复发
     existing = _find_by_trans(db, trans_id)
     if existing is not None:
@@ -83,24 +87,86 @@ def grant_ad_reward(
         rec = AdRewardRecord(
             trans_id=trans_id, user_id=user_id, coin=0, status="capped",
             reward_date=today, reward_name=reward_name, raw=raw,
+            reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm,
         )
         return _commit_record(db, rec, trans_id)
 
-    if coin is None:
-        coin = rewards.get_ad_reward_coin(db)
+    ecpm_raw = ecpm
+    if not ecpm_raw and ad_session_id:
+        ecpm_rec = crud_ecpm.find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+        ecpm_raw = ecpm_rec.ecpm_raw if ecpm_rec is not None else None
+
+    if not ecpm_raw:
+        rec = AdRewardRecord(
+            trans_id=trans_id, user_id=user_id, coin=0, status="ecpm_missing",
+            reward_date=today, reward_name=reward_name, raw=raw,
+            reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=None,
+        )
+        return _commit_record(db, rec, trans_id)
+
+    coin = rewards.calculate_ad_reward_coin(ecpm_raw, _granted_today(db, user_id, today) + 1)
 
     # 发金币 + 记一笔,同事务
     crud_wallet.grant_coins(
         db, user_id, coin,
-        biz_type="ad_reward", ref_id=trans_id, remark="看视频奖励金币",
+        biz_type="reward_video", ref_id=trans_id, remark="看视频奖励金币",
     )
     rec = AdRewardRecord(
         trans_id=trans_id, user_id=user_id, coin=coin, status="granted",
         reward_date=today, reward_name=reward_name, raw=raw,
+        reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm_raw,
     )
     return _commit_record(db, rec, trans_id)
 
 
+def record_external_reward(
+    db: Session,
+    user_id: int,
+    trans_id: str,
+    *,
+    coin: int,
+    reward_scene: str,
+    ad_session_id: str | None = None,
+    ecpm: str | None = None,
+    reward_name: str | None = None,
+    raw: str | None = None,
+    status: str = "granted",
+    commit: bool = True,
+) -> AdRewardRecord:
+    """记录非普通看视频场景的 S2S 回调幂等,发币由调用方业务仓储完成。"""
+    existing = _find_by_trans(db, trans_id)
+    if existing is not None:
+        return existing
+    if db.get(User, user_id) is None:
+        raise UnknownUserError
+    rec = AdRewardRecord(
+        trans_id=trans_id,
+        user_id=user_id,
+        coin=coin,
+        status=status,
+        reward_date=cn_today().isoformat(),
+        reward_name=reward_name,
+        raw=raw,
+        reward_scene=reward_scene,
+        ad_session_id=ad_session_id,
+        ecpm_raw=ecpm,
+    )
+    db.add(rec)
+    if commit:
+        try:
+            db.commit()
+        except IntegrityError:
+            db.rollback()
+            existing = _find_by_trans(db, trans_id)
+            if existing is not None:
+                return existing
+            raise
+        db.refresh(rec)
+    else:
+        db.flush()
+    return rec
+
+
 def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardRecord:
     """提交发奖记录;并发下同 trans_id 撞唯一约束时回滚并返回已存在的那条(幂等兜底)。"""
     db.add(rec)
@@ -125,6 +191,7 @@ def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> li
                 AdRewardRecord.user_id == user_id,
                 AdRewardRecord.reward_date == reward_date,
                 AdRewardRecord.status == "granted",
+                AdRewardRecord.reward_scene == "reward_video",
             )
             .order_by(AdRewardRecord.created_at.desc())
         ).scalars()
@@ -153,7 +220,7 @@ def today_status(
     return (
         len(granted_desc),
         rewards.get_ad_daily_limit(db),
-        rewards.get_ad_reward_coin(db),
+        0,
         state.round_count,
         state.cooldown_until,
         watched_seconds_today(db, user_id, today=today),

app/repositories/comparison.py

@@ -5,9 +5,12 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.core.rewards import CN_TZ
 from app.models.comparison import ComparisonRecord
 from app.models.savings import SavingsRecord
 from app.schemas.compare_record import ComparisonRecordIn
@@ -104,7 +107,15 @@ def upsert_record(
         db.refresh(existing)
         return existing
 
-    rec = ComparisonRecord(user_id=user_id, trace_id=payload.trace_id, **fields)
+    # created_at 显式存 naive 北京 wall-clock(同 savings_record):客户端「比价记录」页
+    # formatRecordTime 原样切片 created_at 字符串、不转时区,而默认 server_default=func.now()
+    # 在 SQLite 下返回 UTC → 直接慢 8h。覆盖分支不动 created_at(保留首次创建时间)。
+    rec = ComparisonRecord(
+        user_id=user_id,
+        trace_id=payload.trace_id,
+        created_at=datetime.now(CN_TZ).replace(tzinfo=None),
+        **fields,
+    )
     db.add(rec)
     db.commit()
     db.refresh(rec)

app/repositories/coupon_state.py

@@ -9,7 +9,7 @@ import logging
 from datetime import date, datetime
 from zoneinfo import ZoneInfo
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -65,6 +65,19 @@ def mark_engagement(
         db.rollback()
 
 
+def reset_today_engagement(db: Session, device_id: str) -> int:
+    """删这台设备今天的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
+    删后 has_engaged_today → false,今天又能弹。返回删除行数。"""
+    result = db.execute(
+        delete(CouponPromptEngagement).where(
+            CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.engage_date == today_cn(),
+        )
+    )
+    db.commit()
+    return result.rowcount or 0
+
+
 # ===== 领券记录(coupon_claim_record)=====
 
 def record_claims(

app/repositories/report.py

@@ -1,9 +1,12 @@
 """price_report 表读写。"""
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from app.core.rewards import CN_TZ
 from app.models.price_report import PriceReport
 
 
@@ -35,6 +38,7 @@ def create_report(
         reported_price_cents=reported_price_cents,
         images=images,
         status="pending",
+        created_at=datetime.now(CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(同 savings/comparison)
     )
     db.add(rep)
     db.commit()

app/repositories/savings.py

@@ -179,6 +179,10 @@ def create_from_report(
         client_event_id=req.client_event_id,
         device_id=req.device_id,
         source="compare",
+        # created_at 显式存 naive 北京 wall-clock(与 demo 行、聚合 _local_date 的 naive 分支一致)。
+        # 不用列默认 server_default=func.now()——SQLite 下它返回 UTC,会让明细页时间早 8 小时。
+        # ⚠️ 生产 PG(timestamptz)对 naive 的解释与 SQLite 不同,迁前端到 PG 时需确认时区一致。
+        created_at=datetime.now(CN_TZ).replace(tzinfo=None),
     )
     db.add(rec)
     db.commit()

app/repositories/signin.py

@@ -32,6 +32,10 @@ class AlreadyBoostedError(Exception):
     """今天签到奖励已经膨胀过。"""
 
 
+class LastCycleDayBoostBlockedError(Exception):
+    """14 天循环最后一天不允许签到膨胀。"""
+
+
 @dataclass
 class SigninStep:
     day: int  # 1..14
@@ -148,13 +152,27 @@ def _today_record(db: Session, user_id: int) -> SigninRecord | None:
     ).scalar_one_or_none()
 
 
+def boost_by_ad_ref(
+    db: Session, user_id: int, ad_ref_id: str
+) -> SigninBoostRecord | None:
+    """按广告交易号查签到膨胀记录。S2S 发奖后客户端确认用。"""
+    return db.execute(
+        select(SigninBoostRecord).where(
+            SigninBoostRecord.user_id == user_id,
+            SigninBoostRecord.ad_ref_id == ad_ref_id,
+        )
+    ).scalar_one_or_none()
+
+
 def boost_today_signin(
-    db: Session, user_id: int, *, ad_ref_id: str | None = None
+    db: Session, user_id: int, *, ad_ref_id: str | None = None, commit: bool = True
 ) -> tuple[SigninBoostRecord, int]:
-    """签到后看广告膨胀:补发一笔等额签到金币。返回 (膨胀记录, 补发后余额)。"""
+    """签到后看广告膨胀:固定补发配置金币。返回 (膨胀记录, 补发后余额)。"""
     record = _today_record(db, user_id)
     if record is None:
         raise NotSignedTodayError
+    if record.cycle_day == SIGNIN_CYCLE_LEN:
+        raise LastCycleDayBoostBlockedError
 
     today = record.signin_date
     existing = db.execute(
@@ -169,17 +187,20 @@ def boost_today_signin(
     boost = SigninBoostRecord(
         user_id=user_id,
         signin_date=today,
-        coin_awarded=record.coin_awarded,
+        coin_awarded=rewards.get_signin_boost_coin(db),
         ad_ref_id=ad_ref_id,
     )
     db.add(boost)
     try:
         acc, _ = crud_wallet.grant_coins(
-            db, user_id, record.coin_awarded,
-            biz_type="signin_boost", ref_id=today.isoformat(),
+            db, user_id, boost.coin_awarded,
+            biz_type="signin_boost", ref_id=ad_ref_id or today.isoformat(),
             remark=f"签到膨胀 第{record.cycle_day}天",
         )
-        db.commit()
+        if commit:
+            db.commit()
+        else:
+            db.flush()
     except IntegrityError as e:
         db.rollback()
         raise AlreadyBoostedError from e

app/repositories/wallet.py

@@ -12,6 +12,7 @@ import uuid
 from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.core import rewards
@@ -31,6 +32,7 @@ from app.models.wallet import (
 _WX_STATE_SUCCESS = "SUCCESS"
 _WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
 _WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM"  # 用户还没在微信确认页确认
+_WITHDRAW_ACTIVE_STATUSES = {"reviewing", "pending"}
 # 免确认收款授权状态
 _WX_AUTH_ACTIVE = "TAKING_EFFECT"   # 已生效,可免确认转账
 _WX_AUTH_CLOSED = "CLOSED"          # 已关闭(用户/商户/风控),需重新开启
@@ -60,6 +62,10 @@ class InsufficientCashError(Exception):
     """现金余额不足。"""
 
 
+class WithdrawTooFrequentError(Exception):
+    """提现申请过于频繁,或已有未完成提现单。"""
+
+
 class WithdrawTransferError(Exception):
     """调用微信转账失败(已退回余额)。"""
 
@@ -116,6 +122,7 @@ def grant_coins(
         biz_type=biz_type,
         ref_id=ref_id,
         remark=remark,
+        created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(客户端原样切片显示)
     )
     db.add(txn)
     db.flush()
@@ -173,6 +180,7 @@ def exchange_coins_to_cash(
             balance_after_cents=acc.cash_balance_cents,
             biz_type="exchange_in",
             remark=remark,
+            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
         )
     )
     db.commit()
@@ -281,6 +289,18 @@ def _refund_withdraw(
     """
     if order.status in ("failed", "rejected"):
         return  # 防重复退款(并发/对账与查单/重复拒绝同时触发)
+    refunded_txn_id = db.execute(
+        select(CashTransaction.id).where(
+            CashTransaction.user_id == order.user_id,
+            CashTransaction.biz_type == "withdraw_refund",
+            CashTransaction.ref_id == order.out_bill_no,
+        ).limit(1)
+    ).scalar_one_or_none()
+    if refunded_txn_id is not None:
+        order.status = final_status
+        order.fail_reason = reason[:256]
+        db.commit()
+        return
     bal = _add_cash(db, order.user_id, order.amount_cents)
     db.add(
         CashTransaction(
@@ -291,11 +311,35 @@ def _refund_withdraw(
             ref_id=order.out_bill_no,
             # 用户可见文案区分"未成功(自动退)"vs"审核未通过";技术原因记在 order.fail_reason
             remark="提现审核未通过,金额已退回" if final_status == "rejected" else "提现未成功,金额已退回",
+            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
         )
     )
     order.status = final_status
     order.fail_reason = reason[:256]
-    db.commit()
+    out_bill_no = order.out_bill_no
+    user_id = order.user_id
+    try:
+        db.commit()
+    except IntegrityError:
+        # 并发退款兜底:唯一退款流水已被另一事务写入时,回滚本事务的加钱和流水,
+        # 再把订单状态补到终态。这样无论拒绝/查单/对账怎么并发,现金最多退一次。
+        db.rollback()
+        refunded_txn_id = db.execute(
+            select(CashTransaction.id).where(
+                CashTransaction.user_id == user_id,
+                CashTransaction.biz_type == "withdraw_refund",
+                CashTransaction.ref_id == out_bill_no,
+            ).limit(1)
+        ).scalar_one_or_none()
+        if refunded_txn_id is None:
+            raise
+        fresh_order = db.execute(
+            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
+        ).scalar_one_or_none()
+        if fresh_order is not None and fresh_order.status not in ("failed", "rejected"):
+            fresh_order.status = final_status
+            fresh_order.fail_reason = reason[:256]
+            db.commit()
 
 
 def _wx_not_found(result: dict) -> bool:
@@ -382,6 +426,15 @@ def create_withdraw(
     else:
         out_bill_no = uuid.uuid4().hex
 
+    active_order_id = db.execute(
+        select(WithdrawOrder.id).where(
+            WithdrawOrder.user_id == user_id,
+            WithdrawOrder.status.in_(_WITHDRAW_ACTIVE_STATUSES),
+        ).limit(1)
+    ).scalar_one_or_none()
+    if active_order_id is not None:
+        raise WithdrawTooFrequentError
+
     # 账户须存在(原子扣款的 UPDATE 不会建账户)
     get_or_create_account(db, user_id, commit=True)
 
@@ -401,6 +454,7 @@ def create_withdraw(
             biz_type="withdraw",
             ref_id=out_bill_no,
             remark="提现到微信零钱(待审核)",
+            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
         )
     )
     order = WithdrawOrder(
@@ -411,7 +465,26 @@ def create_withdraw(
         status="reviewing",
     )
     db.add(order)
-    db.commit()
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        existing = db.execute(
+            select(WithdrawOrder).where(
+                WithdrawOrder.out_bill_no == out_bill_no, WithdrawOrder.user_id == user_id
+            )
+        ).scalar_one_or_none()
+        if existing is not None:
+            return existing
+        active_order_id = db.execute(
+            select(WithdrawOrder.id).where(
+                WithdrawOrder.user_id == user_id,
+                WithdrawOrder.status.in_(_WITHDRAW_ACTIVE_STATUSES),
+            ).limit(1)
+        ).scalar_one_or_none()
+        if active_order_id is not None:
+            raise WithdrawTooFrequentError from None
+        raise
     db.refresh(order)
     return order  # 待管理员审核;**不在此处打款**
 
@@ -699,7 +772,15 @@ def refresh_withdraw_status(
     if order.status != "pending":
         return order  # 已终态,不再查
 
-    result = wxpay.query_transfer(out_bill_no)
+    try:
+        result = wxpay.query_transfer(out_bill_no)
+    except wxpay.WxPayNotConfiguredError:
+        raise
+    except Exception as exc:  # noqa: BLE001 - 查单失败不能把运营后台打成 500
+        order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256]
+        db.commit()
+        db.refresh(order)
+        return order
     if result["status_code"] != 200:
         if _wx_not_found(result):
             # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全

app/schemas/ad.py

@@ -28,7 +28,7 @@ class AdRewardStatusOut(BaseModel):
     used_today: int = Field(..., description="今日已成功发奖次数")
     daily_limit: int = Field(..., description="每日发奖次数上限")
     remaining: int = Field(..., description="今日剩余可领次数")
-    coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
+    coin_per_ad: int = Field(..., description="历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0")
     round_count: int = Field(
         ..., description="本轮已看次数,当前 round_size=1,客户端可由 cooldown_until 判断短冷却"
     )
@@ -46,11 +46,15 @@ class AdRewardStatusOut(BaseModel):
 class EcpmReportIn(BaseModel):
     """客户端上报一次广告展示的 eCPM(内部收益统计/对账)。
 
-    user_id 不在 body 里——由 JWT 取(Bearer),防伪造。ecpm 原样上报字符串(单位待确认)。
+    user_id 不在 body 里——由 JWT 取(Bearer),防伪造。ecpm 原样上报字符串,后端按分/千次展示处理(SDK getEcpm 原值,非元)。
     """
 
     ad_type: str = Field(..., description="广告类型:reward_video(激励视频) / draw(Draw 信息流) 等")
-    ecpm: str = Field(..., description="穿山甲 getShowEcpm().getEcpm() 原始字符串,单位待确认,原样上报")
+    ecpm: str = Field(..., description="穿山甲 getShowEcpm().getEcpm() 原始字符串,按分/千次展示处理(SDK getEcpm 原值,非元)")
+    ad_session_id: str | None = Field(
+        None, min_length=8, max_length=64,
+        description="客户端生成的一次广告会话 id;激励视频 S2S extra 会透传同值",
+    )
     adn: str | None = Field(None, description="实际投放 ADN(getSdkName),如 pangle")
     slot_id: str | None = Field(None, description="实际展示代码位(底层 mediation rit)")
 
@@ -78,16 +82,32 @@ class WatchReportOut(BaseModel):
     watch_seconds_remaining: int = Field(..., description="今日剩余可观看秒数")
 
 
+class TestGrantIn(BaseModel):
+    """[仅本地联调]模拟发奖入参。"""
+
+    reward_scene: str = Field(
+        "reward_video",
+        description="模拟发奖场景:reward_video(普通激励视频) / signin_boost(签到膨胀)",
+    )
+    ad_session_id: str | None = Field(
+        None, min_length=8, max_length=64,
+        description="本次广告会话 id(与 ecpm-report 同值)。reward_video 场景下据此查回客户端"
+        "已上报的真实 eCPM 来按公式发奖;查不到或 eCPM≤0 时兜底 200,保证本地联调仍出非零金币",
+    )
+
+
 class TestGrantOut(BaseModel):
     """[仅本地联调]模拟发奖结果。带上今日进度,客户端可直接据此刷新展示。"""
 
     granted: bool = Field(..., description="本次是否真的发了金币(达每日上限则 False)")
-    status: str = Field(..., description="granted / capped")
+    status: str = Field(
+        ..., description="granted / capped / not_signed / already_boosted / last_day / unknown_scene"
+    )
     coin: int = Field(..., description="本次发放金币(capped 时为 0)")
     used_today: int = Field(..., description="今日已成功发奖次数")
     daily_limit: int = Field(..., description="每日发奖次数上限")
     remaining: int = Field(..., description="今日剩余可领次数")
-    coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
+    coin_per_ad: int = Field(..., description="历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0")
     round_count: int = Field(..., description="本轮已看次数,详见 AdRewardStatusOut")
     cooldown_until: datetime | None = Field(
         None, description="本轮冷却结束时间(UTC),详见 AdRewardStatusOut"
@@ -101,7 +121,10 @@ class FeedRewardIn(BaseModel):
     """
 
     client_event_id: str = Field(..., min_length=8, max_length=64, description="客户端生成的幂等事件 id")
-    ecpm: str = Field(..., description="本条信息流广告 eCPM,按元/千次展示处理")
+    ad_session_id: str | None = Field(
+        None, min_length=8, max_length=64, description="客户端生成的一次信息流广告会话 id"
+    )
+    ecpm: str = Field(..., description="本条信息流广告 eCPM,按分/千次展示处理(SDK getEcpm 原值,非元)")
     duration_seconds: int = Field(..., ge=0, description="本条广告实际展示/播放秒数")
     adn: str | None = Field(None, description="实际投放 ADN")
     slot_id: str | None = Field(None, description="实际展示代码位")

docs/api/README.md

@@ -102,6 +102,7 @@
 | A24 | `DELETE /admin/api/marquee-seeds/{seed_id}` | operator | [详情](./admin-marquee-seeds.md) |
 | A25 | `POST /admin/api/marquee-seeds/bulk` | operator | [详情](./admin-marquee-seeds.md) |
 | A26 | `GET /admin/api/marquee-seeds/preview` | admin | [详情](./admin-marquee-seeds.md) |
+| A27 | `GET /admin/api/ad-coin-audit` | admin | [详情](./admin-ad-coin-audit.md)(看广告金币公式复算对账,只读) |
 | - | `GET /admin/api/health` | 无 | admin 健康检查(无单独文档) |
 
 > ⚠️ 美团三个接口当前**无鉴权**,且 `referral-link` 的 `sid` 允许客户端传值覆盖默认渠道——见各接口"备注"。

docs/api/ad-ecpm-report.md

@@ -8,7 +8,8 @@
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---|---|
 | `ad_type` | str | 是 | 广告类型:`reward_video`(激励视频) / `draw`(Draw 信息流) 等 |
-| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,**单位待确认(分/元)**,原样上报 |
+| `ad_session_id` | str\|null | 否 | 客户端生成的广告会话 ID;需和穿山甲 `extra.ad_session_id` 一致,用于 S2S 缺 eCPM 时匹配 |
+| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,单位是**分/千次展示**(非元),后端 ÷100 转元参与金币公式 |
 | `adn` | str\|null | 否 | 实际投放 ADN(`getSdkName`),如 `pangle` |
 | `slot_id` | str\|null | 否 | 实际展示代码位(底层 mediation rit,非客户端配置位) |
 
@@ -24,7 +25,8 @@
 ## 说明
 客户端在广告**展示后**(`onAdShow` 读 `getShowEcpm()`)调用,把本次展示的 eCPM 落库做**内部收益统计/对账**。
 
-- **best-effort**:客户端 fire-and-forget,丢一两条不影响业务;穿山甲后台报表是结算权威兜底。
+- 普通激励视频发奖会先用 S2S 回调自带 `ecpm`;若缺失,再按 `ad_session_id` 读取本接口上报的 eCPM;两边都没有则不发并记录异常。
+- **best-effort**:客户端 fire-and-forget,但普通激励视频若 S2S 缺 eCPM,这条上报会成为发奖依据。
 - 落 `ad_ecpm_record` 表,`report_date` 用北京时间当天,供「按用户/按天聚合」对账。
 - **与发奖是两条独立流**:发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S,有 `trans_id`、无 ecpm),本接口客户端上报(有 ecpm、无 `trans_id`)。两者无公共键,**不逐条一一对应**,只做按用户/按天聚合口径。
 - eCPM 是**每千次展示预估**,单条展示预估收益 ≈ `ecpm / 1000`,且为客户端预估口径,非最终结算。

docs/api/ad-feed-reward.md

@@ -11,7 +11,8 @@
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---:|---|
 | `client_event_id` | string | 是 | 客户端生成的幂等事件 id，8-64 字符 |
-| `ecpm` | string | 是 | 本条信息流广告 eCPM，按“元/千次展示”处理 |
+| `ad_session_id` | string\|null | 否 | 客户端生成的一次信息流广告会话 id，用于对账/排查 |
+| `ecpm` | string | 是 | 本条信息流广告 eCPM(穿山甲 getEcpm 原值)，按“分/千次展示”处理(非元) |
 | `duration_seconds` | int | 是 | 实际展示/播放秒数 |
 | `adn` | string\|null | 否 | 实际投放 ADN |
 | `slot_id` | string\|null | 否 | 实际展示代码位 |

docs/api/ad-pangle-callback.md

@@ -14,8 +14,8 @@ GroMore 以 GET 回调,关键参数:
 | `user_id` | string | 客户端 `setUserID` 传入的用户标识（须为数字 = 本系统 user.id） |
 | `trans_id` | string | 交易号（**幂等键** + **唯一参与签名的字段**） |
 | `reward_name` | string | 奖励名（广告位配置,入库备注） |
-| `reward_amount` | int | 奖励数量（广告位配置）→ **本次发放金币**。缺/坏/≤0 回退 `AD_REWARD_COIN`,超 `MAX_AD_REWARD_COIN` 夹紧 |
-| `extra` | string | 客户端透传的 customData（可空,入库备注） |
+| `ecpm` | string\|null | GroMore 回调携带的 eCPM。普通激励视频优先用它计算金币 |
+| `extra` / `gromoreExtra` / `gromore_extra` | string | 客户端透传 JSON。支持 `ad_session_id`、`reward_scene`;`reward_scene=signin_boost` 表示签到膨胀 |
 | `mediation_rit` | string | 代码位 ID（GroMore 带,目前仅入 raw 备查） |
 | `prime_rit` | string | 广告位 ID（同上） |
 | `adn_name` | string | 实际出广告的 ADN 名（同上,可用于收益分析） |
@@ -37,4 +37,9 @@ GroMore 以 GET 回调,关键参数:
 - `503` 回调未配置（`pangle_callback_configured=false`）
 
 ## 说明
-**发奖唯一可信入口**:验签 → 取 `user_id` → 按 `reward_amount` 解析金币 → 幂等发金币（按 `trans_id` 去重 + 每日上限）。客户端不参与发奖,被破解也刷不到钱。验签过但参数缺/坏或 user 不存在 → 不发（`is_verify=false` + `reason`）;granted / capped → `is_verify=true` + `reason=0`。
+**发奖唯一可信入口**:验签 → 取 `user_id`/`extra` → 按 `reward_scene` 分流 → 幂等处理（按 `trans_id` 去重）。客户端不直接发奖,被破解也刷不到钱。
+
+- `reward_scene=reward_video` 或缺省:普通激励视频。金币按 `eCPM / 1000 * eCPM因子 * 当日次数因子 * 10000` 计算;若回调没有 `ecpm`,会按 `extra.ad_session_id` 查客户端 `/ad/ecpm-report` 的上报值;两边都没有 eCPM 时不发币,记录 `status=ecpm_missing`。
+- `reward_scene=signin_boost`:签到膨胀。要求用户当天已签到且不是 Day14;看完视频固定发 `2000` 金币,写 `signin_boost_record` 与 `coin_transaction.biz_type=signin_boost`。
+- 未知 `reward_scene`:不发币,记录 `status=unknown_scene`,返回 `is_verify=false/reason=1`。
+- 验签过但参数缺/坏或 user 不存在 → 不发（`is_verify=false` + `reason`）;granted / capped / ecpm_missing / 业务不满足已记录 → `is_verify=true` + `reason=0`。

docs/api/ad-reward-status.md

@@ -13,7 +13,7 @@
 | `used_today` | int | 今日已成功发奖次数 |
 | `daily_limit` | int | 每日发奖次数上限 |
 | `remaining` | int | 今日剩余可领次数 |
-| `coin_per_ad` | int | 看完一个激励视频发的金币 |
+| `coin_per_ad` | int | 历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0 |
 | `round_count` | int | 本轮已看次数;当前 `round_size=1`,广告关闭后进入短冷却 |
 | `cooldown_until` | datetime\|null | 3 秒短冷却结束时间(UTC ISO);null 表示不在冷却 |
 | `watched_seconds_today` | int | 今日已上报的激励视频观看秒数;当前仅兼容/排查用 |
@@ -22,7 +22,7 @@
 
 ## 说明
 福利页「看视频赚金币」用:
-- 展示「今日还能看 N 次」「看一次得 M 金币」(`remaining`/`coin_per_ad`)
+- 展示「今日还能看 N 次」(`remaining`);单次金币正式按 eCPM 动态计算,不再用 `coin_per_ad` 预估
 - 任务行 CTA 4 态推导:`adLoading` → Loading;`remaining==0` → Capped("明天再来");`cooldown_until` 在未来 → CoolingDown(显示 MM:SS 倒计时,不可点);否则 Normal("去赚取")
 - 弹窗 limit note:`used_today >= daily_limit` 显示「今日视频已到限额，明天再来」;`cooldown_until!=null` 显示「广告冷却中，3秒后再来」
 

docs/api/ad-test-grant.md

@@ -5,7 +5,12 @@
 > ⚠️ **仅本地联调**,受 `AD_REWARD_TEST_GRANT_ENABLED` 开关控制,**生产必须关闭**（默认 False → 一律 404）。
 
 ## 入参
-无（用户由 token 确定）。
+请求体可省略；用户由 token 确定。
+
+| 字段 | 类型 | 必填 | 默认 | 说明 |
+|---|---|---|---|---|
+| `reward_scene` | string | 否 | `reward_video` | 模拟发奖场景。`reward_video`=普通激励视频；`signin_boost`=签到膨胀 |
+| `ad_session_id` | string(8~64) \| null | 否 | null | 本次广告会话 id(与 [ecpm-report](./ad-ecpm-report.md) 同值)。**仅 `reward_video` 场景生效**:据此查回客户端已上报的真实 eCPM,走与正式发奖相同的公式发奖;查不到或 eCPM≤0(测试应用常返 0/假值)时兜底 200,保证本地联调仍出非零金币 |
 
 ## 出参
 响应 `200`:`TestGrantOut`
@@ -13,15 +18,19 @@
 | 字段 | 类型 | 说明 |
 |---|---|---|
 | `granted` | bool | 本次是否真发了金币（达每日上限则 false） |
-| `status` | string | `granted` / `capped`（达上限） |
+| `status` | string | `granted` / `capped`（达上限） / `not_signed` / `already_boosted` / `last_day` / `unknown_scene` |
 | `coin` | int | 本次发放金币（capped 时 0） |
 | `used_today` | int | 今日已成功发奖次数 |
 | `daily_limit` | int | 每日发奖次数上限 |
 | `remaining` | int | 今日剩余可领次数 |
-| `coin_per_ad` | int | 看完一个激励视频发的金币 |
+| `coin_per_ad` | int | 历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0 |
 
 ## 错误码
 - `404` 开关未开（伪装不存在） / 用户不存在
 
 ## 说明
-没公网、穿山甲 S2S 回调打不到本地时,debug 客户端看完广告后调它,直接走与 [ad-pangle-callback](./ad-pangle-callback.md) 相同的发奖逻辑（每次新 `trans_id`,幂等 + 每日上限）。它让已登录客户端能自助发奖 = 绕过反作弊,**严禁在生产开启**。
+没公网、穿山甲 S2S 回调打不到本地时,debug 客户端看完广告后调它,直接走与 [ad-pangle-callback](./ad-pangle-callback.md) 相同的发奖逻辑（每次新 `trans_id`,幂等 + 每日上限/今日膨胀一次）。
+
+`reward_scene=reward_video` 时按上面 `ad_session_id` 查回的真实 eCPM 走金币公式发奖(取不到兜底 200)——便于本地用 [admin 金币审计](./admin-ad-coin-audit.md) 核对「看广告→金币」是否按公式计算。
+
+`reward_scene=signin_boost` 时复用签到膨胀业务规则:必须当天已签到、非第 14 天、当天未膨胀过,成功后写入 `signin_boost` 金币流水。它让已登录客户端能自助发奖 = 绕过反作弊,**严禁在生产开启**。

docs/api/admin-ad-coin-audit.md

@@ -0,0 +1,69 @@
+# Admin 看广告金币审计
+
+> 所属:Admin 组(前缀 `/admin/api/ad-coin-audit`) | 鉴权:Admin Bearer(任意已登录 admin,只读) | [← 返回 API 索引](./README.md)
+
+把「看视频赚金币」(`ad_reward_record`)和「比价信息流广告」(`ad_feed_reward_record`)两类发奖记录,用与**正式发奖完全相同**的公式 [`app/core/rewards.py` `calculate_ad_reward_coin`](../../app/core/rewards.py) 复算一遍 `expected_coin`,与实际入账的 `actual_coin` 对比,核对金币公式是否生效。**纯只读对账**,不发币、不改任何数据。
+
+相关表:[ad_reward_record](../database/ad_reward_record.md)、[ad_feed_reward_record](../database/ad_feed_reward_record.md)。
+
+## 金币公式(展示参照)
+
+```
+eCPM元 = getEcpm分 ÷ 100
+单份金币 = round( eCPM元 ÷ 1000 × 因子1 × 因子2 × coin_per_yuan )   # coin_per_yuan=10000
+```
+- **eCPM 口径**:`getEcpm()` 原值单位是分/千次展示;先 ÷100 转元,**因子判档与收益换算都用元**。
+- **因子1(eCPM 档,阈值单位=元/千次)**:≤100→0.1,101–200→0.3,201–400→0.4,>400→0.6(即 ¥100/¥200/¥400 CPM 分界。**真实 eCPM 多 <¥100 CPM,故常落最低档 0.1,高档基本不触发——产品有意取舍**)。
+- **因子2(LT,当日第 N 份)**:第1→2.0,第2→1.5,第3→1.3,第4–10→1.1,≥11→1.0。
+- **第 N 份的计数**:看视频每条 granted 记 1 份;信息流每满 10 秒记 1 份。**两个点位各自独立计数**(看视频的份数不影响信息流的份数,反之亦然)。
+
+## GET /admin/api/ad-coin-audit — 复算对比
+
+- 入参(均 query,可选):
+  | 参数 | 类型 | 默认 | 说明 |
+  |---|---|---|---|
+  | `date` | string | 今天 | 北京时间 `YYYY-MM-DD`,审计某天 |
+  | `user_id` | int | 全部 | 只看某用户;不传=所有用户 |
+  | `scene` | string | 两类 | `reward_video` / `feed`;不传=两类都返回 |
+  | `limit` | int(1~500) | 100 | 返回明细条数(按时间倒序截断;**份序号在截断前已按全天数据算好**,不影响复算正确性) |
+
+- 出参 `200`:`AdCoinAuditOut`
+  | 字段 | 类型 | 说明 |
+  |---|---|---|
+  | `date` | string | 审计日期 |
+  | `formula` | object | 当前公式参数快照(见下) |
+  | `total` | int | 返回明细条数 |
+  | `mismatch_count` | int | 其中 `matched=false` 的条数;**=0 说明全部按公式发放** |
+  | `items` | `AdCoinAuditRow[]` | 明细(见下) |
+
+### AdCoinFormulaOut(`formula`)
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `description` | string | 公式文字说明 |
+| `coin_per_yuan` | int | 金币:元 汇率(10000) |
+| `ecpm_unit` | string | eCPM 口径(分/千次展示,SDK getEcpm 原值) |
+| `feed_unit_seconds` | int | 信息流每多少秒折 1 份(10) |
+| `ecpm_factor_tiers` | `[因子, 下限, 上限\|null][]` | 因子1 档位表(直接读发奖常量,与发奖同源) |
+| `lt_factor_tiers` | `[因子, 下限, 上限\|null][]` | 因子2 LT 档位表 |
+
+### AdCoinAuditRow(`items[]`)
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `scene` | string | `reward_video` / `feed` |
+| `record_id` | int | 对应记录表主键 |
+| `user_id` | int | |
+| `created_at` | datetime | |
+| `status` | string | `granted` / `capped`(超每日上限未发) / `ecpm_missing`(缺 eCPM 未发) |
+| `ecpm` | string \| null | 本次采用的 eCPM 原始值 |
+| `ecpm_factor` | float \| null | 因子1;非 granted 为 null |
+| `units` | int | 折算份数:看视频恒 1;信息流 = 满 10 秒份数 |
+| `lt_index_start` / `lt_index_end` | int \| null | 本条占用「当日第几份」的起止(看视频起=止;信息流一条可跨多份) |
+| `lt_factor_start` / `lt_factor_end` | float \| null | 因子2 的起止值(信息流跨份时会衰减,故给区间) |
+| `expected_coin` | int | 按公式复算应发金币(非 granted 恒 0) |
+| `actual_coin` | int | 实际入账金币 |
+| `matched` | bool | 复算与实发是否一致;非 granted 校验实发是否确为 0 |
+
+## 说明
+- **非 granted 行**(capped/ecpm_missing)不占用份序号、应发恒 0,`matched` 用于校验「该不发的确实没发」。
+- 用法建议:**按 `user_id`+`date` 定位某次具体核对**最直观;不带 `user_id` 是全用户当天概览。
+- 本地联调造数:debug 包看完激励视频会调 [`/api/v1/ad/test-grant`](./ad-test-grant.md),它会用客户端按 `ad_session_id` 上报的真实 eCPM 发奖(取不到兜底 200),所以本审计能看到真实 eCPM 对应的金币。

docs/api/admin-stats-overview.md

@@ -31,6 +31,14 @@
 | 字段 | 类型 | 说明 |
 |---|---|---|
 | `granted_total` | int | 累计发放金币(coin_transaction 中所有 `amount > 0` 之和;负数兑换/扣减不计) |
+| `reward_video_coin_total` | int | 普通激励视频累计发放金币(`biz_type=reward_video/ad_reward`) |
+| `reward_video_watch_count` | int | 普通激励视频有效发奖视频数(`ad_reward_record.reward_scene=reward_video,status=granted`) |
+| `feed_ad_coin_total` | int | 信息流广告累计发放金币(`biz_type=feed_ad_reward`) |
+| `feed_ad_watch_count` | int | 信息流广告有效完成视频数(`ad_feed_reward_record.status=granted`) |
+| `signin_coin_total` | int | 签到累计发放金币(`biz_type=signin`) |
+| `signin_count` | int | 签到次数(`signin_record`) |
+| `signin_boost_coin_total` | int | 签到膨胀累计发放金币(`biz_type=signin_boost`) |
+| `signin_boost_watch_count` | int | 签到膨胀有效视频数(`signin_boost_record`) |
 
 **DashboardCash**
 | 字段 | 类型 | 说明 |
@@ -64,6 +72,6 @@
 ## 说明
 - 全部为全局只读聚合(count / sum / DAU / 成功率),不改任何数据。
 - `new_today` / `dau` 按**北京时区**(UTC+8)切天,其余金额/计数无时区概念。
-- 金额单位:`*_cents` 为分;金币(`granted_total`)为个数。
+- 金额单位:`*_cents` 为分;金币(`*_coin_total`/`granted_total`)为个数。
 - CPS 收入数据源未接入(referral-link 只换链接,转化/佣金未回收),`cps` 恒为 `{available:false, note:...}`,前端显示"待接入"。
 - 关联表:[user](../database/user.md) / [coin_transaction](../database/coin_transaction.md) / [withdraw_order](../database/withdraw_order.md) / [comparison_record](../database/comparison_record.md) / [feedback](../database/feedback.md)。

docs/api/signin-boost.md

@@ -1,6 +1,6 @@
 # POST /api/v1/signin/boost — 签到后看广告膨胀金币
 
-用户当天已签到后，看完一条激励视频，补发一笔等额签到金币。
+用户 Day1-Day13 当天已签到后，看完一条激励视频，由穿山甲 S2S 回调固定补发 2000 金币。本接口只用于 S2S 发奖后的确认。
 
 ## 鉴权
 
@@ -10,7 +10,7 @@
 
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---:|---|
-| `ad_ref_id` | string\|null | 否 | 广告会话/交易号。当前开发期可空，后续接 S2S 后可回填 |
+| `ad_ref_id` | string | 是 | 穿山甲 S2S 回调的 `trans_id`。回调需先以 `extra.reward_scene=signin_boost` 完成发奖 |
 
 ## 响应
 
@@ -23,10 +23,11 @@
 ## 错误
 
 - `401`: 未登录
-- `409`: 当天未签到，或当天已经膨胀过
+- `409`: 缺少/无效广告回调记录，非本人广告，回调未发奖，当天未签到，Day14，或当天已经膨胀过
 
 ## 数据写入
 
-- `signin_boost_record` 新增一行，用 `(user_id, signin_date)` 唯一约束防重复。
-- `coin_account` 增加余额。
-- `coin_transaction` 写入 `biz_type=signin_boost`。
+- 本接口不直接发奖;实际写入发生在 `/ad/pangle-callback` 的 `reward_scene=signin_boost` 分支。
+- 回调写 `signin_boost_record` 新增一行，用 `(user_id, signin_date)` 唯一约束防重复。
+- 回调使 `coin_account` 增加固定 `2000` 金币。
+- 回调写入 `coin_transaction.biz_type=signin_boost`。

docs/database/OVERVIEW.md

@@ -64,7 +64,7 @@
 | 金币兑现金 `POST /wallet/exchange` | `coin_account`(U) + `coin_transaction`(C `exchange_out` −) + `cash_transaction`(C `exchange_in` +) | 同事务 |
 | 发起提现 `POST /wallet/withdraw` | `withdraw_order`(C `reviewing`) + `coin_account`(U 扣现金) + `cash_transaction`(C `withdraw` −) | 同事务,**不打款** |
 | 查提现状态 / 用户取消 `GET /wallet/withdraw/status` | `withdraw_order`(U) + 失败→`cash_transaction`(C `withdraw_refund` +) | |
-| 穿山甲发奖 S2S 回调 `POST /ad/pangle-callback` | `ad_reward_record`(C)+ granted→`coin_account`(U)+`coin_transaction`(C `ad_reward`) | `trans_id` 幂等 |
+| 穿山甲发奖 S2S 回调 `POST /ad/pangle-callback` | `ad_reward_record`(C)+ granted→`coin_account`(U)+`coin_transaction`(C `reward_video`/`signin_boost`) | `trans_id` 幂等 |
 | 看广告时长上报 `POST /ad/watch-report` | `ad_watch_log`(C) | |
 | 广告 eCPM 上报 `POST /ad/ecpm-report` | `ad_ecpm_record`(C) | |
 | 信息流广告结算 `POST /ad/feed-reward` | `ad_feed_reward_record`(C)+ granted→`coin_account`(U)+`coin_transaction`(C `feed_ad_reward`) | `client_event_id` 幂等 |
@@ -105,7 +105,7 @@
   | `signin` | 当天日期串(= `signin_record.signin_date` 的 ISO `YYYY-MM-DD`) | + |
   | `signin_boost` | 当天日期串(= `signin_boost_record.signin_date` 的 ISO `YYYY-MM-DD`) | + |
   | `task_<key>` | `user_task.task_key` | + |
-  | `ad_reward` | `ad_reward_record.trans_id` | + |
+  | `reward_video` / `ad_reward`(历史) | `ad_reward_record.trans_id` | + |
   | `feed_ad_reward` | `ad_feed_reward_record.client_event_id` | + |
   | `exchange_out` | null(兑现金,无单据) | − |
   | `admin_grant` / `admin_deduct` | null(原因记在 `remark`=`admin:<reason>`) | + / − |
@@ -118,7 +118,7 @@
   | `exchange_in` | null | + |
 
 - **`comparison_record.store_name` ≈ `savings_record.shop_name`**:无 id 关联,按**店名字符串相等**给比价记录打「已下单」标记(瞬态,不写库)。两边店名同源 = 比价意图识别阶段的门店 query,语义=**店级**(同店比价多次会一并标已下单)。
-- **广告流互不关联**:`ad_reward_record` / `ad_watch_log` / `ad_ecpm_record` / `ad_feed_reward_record` 之间**无公共键**,各自只按 `(user_id, 日期串)` 聚合。别试图 join 它们逐条对应。
+- **广告流会话关联**:`ad_reward_record.ad_session_id` 可与 `ad_ecpm_record.ad_session_id` 对齐;`ad_watch_log` 仍是旧版兼容统计,不逐条参与发奖。
 - **里程碑解锁进度不存库**:`comparison_milestone_claim` 只记「哪几档已领」;进度 = `comparison_record` 里 `status='success'` 的 `count`。
 
 ### ER 关系(文字版)

docs/database/ad_ecpm_record.md

@@ -2,7 +2,7 @@
 
 > 模型 `app/models/ad_ecpm.py` · 仓库 `app/repositories/ad_ecpm.py` · 接口 [ad-ecpm-report](../api/ad-ecpm-report.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
 
-每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM。看广告三条数据流之一(**收益对账**),与 `ad_reward`(发奖,有 trans_id 无 ecpm)、`ad_watch_log`(时长)并列,**无公共键**——只用于**按用户/按天聚合**估算内部广告收益,不做"这条发奖 = 这条展示"的逐条精确关联。穿山甲后台报表才是结算权威,本表是细粒度补充。
+每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM。`ad_session_id` 可与 S2S 回调 `extra.ad_session_id` 对齐:当普通激励视频回调没有 `ecpm` 时,后端会用它匹配本表作为发奖 eCPM 依据。
 
 ## 用在哪 / 增删改查
 - **C(插入)**:`POST /ad/ecpm-report`(`create_ecpm_record`)。每次广告展示上报一条;best-effort,丢一两条不影响业务。鉴权接口已确保 user 存在。
@@ -15,18 +15,19 @@
 | `id` | Integer | PK, autoincrement | |
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
 | `ad_type` | String(32) | NOT NULL | 广告类型,取值如 `reward_video`(激励视频)/ `draw`(Draw 信息流);各类型各自上报,不强行统一代码位 |
+| `ad_session_id` | String(64) | UNIQUE, index, nullable | 客户端广告会话 ID;用于普通激励视频在 S2S 缺 `ecpm` 时匹配发奖 |
 | `adn` | String(32) | nullable | 实际投放 ADN(`getShowEcpm().getSdkName()`,如 `pangle`/`gdt`) |
 | `slot_id` | String(64) | nullable | 实际展示用代码位(底层 mediation rit,非客户端配置位) |
-| `ecpm_raw` | String(32) | NOT NULL | 客户端上报的 eCPM **原始串**(单位待确认,原样存) |
+| `ecpm_raw` | String(32) | NOT NULL | 客户端上报的 eCPM **原始串**(穿山甲 getEcpm 原值,单位**分/千次展示**);后端 ÷100 转元参与金币公式 |
 | `report_date` | String(10) | index, NOT NULL | 北京时间日期串 `YYYY-MM-DD`,按它做按天聚合 |
 | `created_at` | DateTime(tz) | server_default now(), index | 时间 |
 
 ## 关系 / Join Key
 - `user_id` → `user.id`(多对一)。
-- 与 `ad_reward_record` / `ad_watch_log` **无公共键**(独立数据流),只按 `(user_id, report_date)` 聚合。
+- `ad_session_id` → 可关联 `ad_reward_record.ad_session_id`。
 
 ## 索引与约束
-- PK `id`;index `user_id`、`report_date`、`created_at`。
+- PK `id`;index `user_id`、`report_date`、`created_at`、`ad_session_id`;UNIQUE `ad_session_id`。
 
 ## 注意
-- ⚠️ `ecpm_raw` 单位(分/元)截至 2026-05-31 未最终确认;确认后再加一列解析好的数值,在此之前对账按"待定单位"处理。
+- S2S 回调自带 `ecpm` 优先;缺失时才使用同 `ad_session_id` 的客户端上报值。两者都没有时普通激励视频不发币,`ad_reward_record.status=ecpm_missing`。

docs/database/ad_feed_reward_record.md

@@ -8,6 +8,7 @@
 |---|---|---|---|
 | `id` | Integer | PK | 自增主键 |
 | `client_event_id` | String(64) | UNIQUE, NOT NULL | 客户端幂等事件 id |
+| `ad_session_id` | String(64) | index, nullable | 客户端生成的一次信息流广告会话 id |
 | `user_id` | Integer | FK → `user.id`, index, NOT NULL | 用户 |
 | `reward_date` | String(10) | index, NOT NULL | 北京时间日期 `YYYY-MM-DD` |
 | `duration_seconds` | Integer | NOT NULL | 实际展示/播放秒数 |

docs/database/ad_reward_record.md

@@ -2,10 +2,10 @@
 
 > 模型 `app/models/ad_reward.py` · 仓库 `app/repositories/ad_reward.py` · 接口 [ad-pangle-callback](../api/ad-pangle-callback.md) / [ad-reward-status](../api/ad-reward-status.md) / [ad-test-grant](../api/ad-test-grant.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
 
-每条 = 穿山甲一次**服务端发奖回调**(用户看完激励视频)。是看广告三条数据流之一(**发奖**;另两条:`ad_watch_log` 时长、`ad_ecpm_record` 收益,三者无公共键)。`trans_id` 唯一做幂等键(穿山甲会重试,同号只发一次)。`reward_date`(北京时间日期串)给"每日上限"计数用。
+每条 = 穿山甲一次**服务端激励回调**。`trans_id` 唯一做幂等键(穿山甲会重试,同号只处理一次)。`reward_scene` 区分普通激励视频、签到膨胀等场景;`reward_date`(北京时间日期串)给普通激励视频"每日上限"计数用。
 
 ## 用在哪 / 增删改查
-- **C(插入)**:`POST /ad/pangle-callback`(穿山甲 S2S,经 SHA256 验签;`grant_ad_reward`)或 `POST /ad/test-grant`(本地联调)。三道闸:① 验签不过 → API 层 403,不进库;② `trans_id` 已存在 → 原样返回不重复发;③ **当日发奖次数(`DAILY_AD_REWARD_LIMIT`,默认 500)到顶** → 记一行 `status='capped'`、`coin=0`、不发币。旧的观看总时长闸仅在 `DAILY_AD_WATCH_SECONDS_LIMIT > 0` 时启用,当前默认停用。否则 `granted` + `grant_coins(biz_type='ad_reward', ref_id=trans_id)` 加币,同事务。
+- **C(插入)**:`POST /ad/pangle-callback`(穿山甲 S2S,经 SHA256 验签;`grant_ad_reward` 或场景业务处理)或 `POST /ad/test-grant`(本地联调)。普通激励视频三道闸:① 验签不过 → API 层 403,不进库;② `trans_id` 已存在 → 原样返回不重复发;③ **当日发奖次数(`DAILY_AD_REWARD_LIMIT`,默认 500)到顶** → 记一行 `status='capped'`、`coin=0`、不发币。否则按 eCPM 公式发币。
 - **U / D**:无。
 - **R**:`GET /ad/reward-status`(看广告页:今日已发次数/上限、单次金币、本轮已看/冷却结束、今日已看时长/上限);审计/对账整表回溯。
 
@@ -13,10 +13,13 @@
 | 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
 |---|---|---|---|
 | `id` | Integer | PK, autoincrement | |
-| `trans_id` | String(64) | UNIQUE, index, NOT NULL | 穿山甲交易号(幂等键)。**被 `coin_transaction.ref_id` 引用**(biz_type=ad_reward) |
+| `trans_id` | String(64) | UNIQUE, index, NOT NULL | 穿山甲交易号(幂等键)。**被 `coin_transaction.ref_id` 引用**(biz_type=reward_video/signin_boost 等) |
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户(回调 media_extra 带回;不存在抛 UnknownUserError) |
-| `coin` | Integer | NOT NULL, default 0 | 实发金币;`capped` 时为 0 |
-| `status` | String(16) | NOT NULL, default `granted` | 取值:`granted`(已发)/ `capped`(当日次数超限,记录但不发) |
+| `reward_scene` | String(32) | NOT NULL, default `reward_video` | 奖励场景:`reward_video` 普通激励视频;`signin_boost` 签到膨胀 |
+| `ad_session_id` | String(64) | index, nullable | 客户端广告会话 ID,来自 `extra.ad_session_id`;用于匹配 `ad_ecpm_record` |
+| `ecpm_raw` | String(32) | nullable | 本次发奖采用的 eCPM 原始值;可来自 S2S `ecpm` 或客户端上报 |
+| `coin` | Integer | NOT NULL, default 0 | 实发金币;`capped`/`ecpm_missing`/业务不满足时为 0 |
+| `status` | String(16) | NOT NULL, default `granted` | 取值:`granted`(已发)/ `capped`(当日次数超限)/ `ecpm_missing`(缺 eCPM)/ `not_signed`/`already_boosted`/`last_day` |
 | `reward_date` | String(10) | index, NOT NULL | 北京时间日期串 `YYYY-MM-DD`,按它等值统计当日发奖次数 |
 | `reward_name` | String(64) | nullable | 穿山甲上报奖励名(参考,不作发奖依据) |
 | `raw` | String(1024) | nullable | 回调原始参数(审计排查) |
@@ -24,12 +27,13 @@
 
 ## 关系 / Join Key
 - `user_id` → `user.id`(多对一)。
-- `trans_id` ← 被 `coin_transaction.ref_id` 引用(`granted` 那条发币流水);`capped` 行不发币、无对应流水。
-- 与 `ad_watch_log` / `ad_ecpm_record` **无公共键**;当前不逐条关联 eCPM 和发奖回调。
+- `trans_id` ← 被 `coin_transaction.ref_id` 引用(`granted` 那条发币流水);未发币状态无对应流水。
+- `ad_session_id` → 可关联 `ad_ecpm_record.ad_session_id`。
 
 ## 索引与约束
-- PK `id`;UNIQUE+index `trans_id`;index `user_id`、`reward_date`、`created_at`。
+- PK `id`;UNIQUE+index `trans_id`;index `user_id`、`reward_date`、`created_at`、`ad_session_id`。
 
 ## 注意
-- 当前公网回调尚未接入前,激励视频精确 eCPM 绑定发奖先搁置;实发金币仍以穿山甲回调 `reward_amount` 为准(`resolve_ad_reward_coin` 解析,缺/坏回退默认、超上限夹紧),单次金币/每日上限/单次上限均从 `app_config` 读(运营后台可改)。
+- 普通激励视频按 eCPM 公式发奖;若 S2S 与客户端会话上报都缺 eCPM,记录 `status='ecpm_missing'`、`coin=0`,不发币。
+- 签到膨胀复用本表记录 S2S 幂等,实发固定 `2000` 金币由 `signin_boost_record`/`coin_transaction.biz_type=signin_boost` 承载。
 - 并发同 `trans_id` 撞唯一约束 → catch IntegrityError 回滚返回已存在那条(幂等兜底)。

docs/database/coin_transaction.md

@@ -12,7 +12,7 @@
   | 签到 `POST /signin/do` | `signin` | + | 当天日期串(= `signin_record.signin_date` ISO) |
   | 签到后看广告膨胀 `POST /signin/boost` | `signin_boost` | + | 当天日期串(= `signin_boost_record.signin_date` ISO) |
   | 领任务 `POST /tasks/claim` | `task_<key>`(如 `task_enable_notification`) | + | `user_task.task_key` |
-  | 看广告发奖回调 `POST /ad/pangle-callback` | `ad_reward` | + | `ad_reward_record.trans_id` |
+  | 普通激励视频 S2S 回调 `POST /ad/pangle-callback` | `reward_video`(历史兼容:`ad_reward`) | + | `ad_reward_record.trans_id` |
   | 信息流广告结算 `POST /ad/feed-reward` | `feed_ad_reward` | + | `ad_feed_reward_record.client_event_id` |
   | 金币兑现金 `POST /wallet/exchange` | `exchange_out` | − | null(配套 `cash_transaction.exchange_in`) |
   | admin 手动加金币 | `admin_grant` | + | null(`remark`=`admin:<reason>`) |
@@ -29,14 +29,14 @@
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
 | `amount` | Integer | NOT NULL | 本笔变动金币;**正=入账(赚),负=出账(花/兑换)** |
 | `balance_after` | Integer | NOT NULL | 本笔后金币余额(= 当时 `coin_account.coin_balance`,对账用) |
-| `biz_type` | String(32) | NOT NULL | 取值见上表:`signin` / `signin_boost` / `task_<key>` / `ad_reward` / `feed_ad_reward` / `exchange_out` / `admin_grant` / `admin_deduct`。无 DB 枚举约束,靠写入方约定 |
+| `biz_type` | String(32) | NOT NULL | 取值见上表:`signin` / `signin_boost` / `task_<key>` / `reward_video` / `ad_reward`(历史) / `feed_ad_reward` / `exchange_out` / `admin_grant` / `admin_deduct`。无 DB 枚举约束,靠写入方约定 |
 | `ref_id` | String(64) | nullable | **关联业务键,指向随 `biz_type` 变(见上表)**;无关联时为 null |
 | `remark` | String(128) | nullable | 备注(如签到「每日签到 第N天」、admin「admin:<reason>」) |
 | `created_at` | DateTime(tz) | server_default now(), index | 时间 |
 
 ## 关系 / Join Key
 - `user_id` → `user.id`(多对一)。
-- `ref_id` 是**软关联**(无 FK),目标表随 `biz_type`:`signin`→签到日 / `signin_boost`→签到日 / `task_<key>`→`user_task.task_key` / `ad_reward`→`ad_reward_record.trans_id` / `feed_ad_reward`→`ad_feed_reward_record.client_event_id` / 其余 null。
+- `ref_id` 是**软关联**(无 FK),目标表随 `biz_type`:`signin`→签到日 / `signin_boost`→`ad_reward_record.trans_id` / `task_<key>`→`user_task.task_key` / `reward_video`/`ad_reward`→`ad_reward_record.trans_id` / `feed_ad_reward`→`ad_feed_reward_record.client_event_id` / 其余 null。
 
 ## 索引与约束
 - PK `id`;index `user_id`、`created_at`。

docs/database/signin_boost_record.md

@@ -1,6 +1,6 @@
 # signin_boost_record — 签到膨胀记录
 
-App 用户当天签到后，看完激励视频可把签到奖励膨胀一次。本表记录膨胀动作，并用唯一约束防重复补发。
+App 用户 Day1-Day13 当天签到后，看完激励视频可固定膨胀 2000 金币一次。Day14 不展示也不允许膨胀。本表记录膨胀动作，并用唯一约束防重复补发。
 
 ## 字段
 
@@ -9,8 +9,8 @@ App 用户当天签到后，看完激励视频可把签到奖励膨胀一次。
 | `id` | Integer | PK | 自增主键 |
 | `user_id` | Integer | FK → `user.id`, index, NOT NULL | 用户 |
 | `signin_date` | Date | NOT NULL | 被膨胀的签到日期，北京时间 |
-| `coin_awarded` | Integer | NOT NULL | 本次补发金币 |
-| `ad_ref_id` | String(64) | nullable | 广告会话/交易号，开发期可空 |
+| `coin_awarded` | Integer | NOT NULL | 本次补发金币,默认固定 2000 |
+| `ad_ref_id` | String(64) | nullable | 穿山甲 S2S 回调 `trans_id` |
 | `created_at` | DateTime | NOT NULL | 创建时间 |
 
 ## 约束
@@ -19,4 +19,4 @@ App 用户当天签到后，看完激励视频可把签到奖励膨胀一次。
 
 ## 关联
 
-- 膨胀成功时写 `coin_transaction.biz_type=signin_boost`，`ref_id=signin_date`。
+- 膨胀成功时写 `coin_transaction.biz_type=signin_boost`，`ref_id=ad_ref_id`。

docs/integrations/pangle.md

@@ -9,7 +9,7 @@
 **`sign = SHA256("{m-key}:{trans_id}")` 的十六进制串。**
 
 - 是**普通 SHA256**，不是 HMAC，也不是 RSA。
-- **只对 `m-key:trans_id` 这一个字符串**签名，`user_id` / `reward_amount` 等其余参数**不参与**签名。它们的可信度由"能算出正确 sign = 知道 m-key"间接保证——伪造者没有 m-key 连合法 sign 都造不出，自然无法注入假 `user_id`。
+- **只对 `m-key:trans_id` 这一个字符串**签名，`user_id` / `ecpm` / `extra` 等其余参数**不参与**签名。它们的可信度由"能算出正确 sign = 知道 m-key"间接保证——伪造者没有 m-key 连合法 sign 都造不出，自然无法注入假 `user_id`。
 - 这是穿山甲 **GroMore「广告位 → 服务端激励回调」官方规范**（supportcenter/26240）。我们客户端是 `useMediation(true)` 融合，回调走 **GroMore 广告位层级**，不是联盟代码位层级。
 
 ## 函数

tests/test_ad_reward.py

@@ -10,9 +10,7 @@ from sqlalchemy import select
 from app.core.config import settings
 from app.integrations import pangle
 from app.core.rewards import (
-    AD_REWARD_COIN,
     DAILY_AD_REWARD_LIMIT,
-    MAX_AD_REWARD_COIN,
     VIDEO_ROUND_COOLDOWN_SECONDS,
     VIDEO_ROUND_REQUIRED_COUNT,
     calculate_ad_reward_coin,
@@ -57,41 +55,34 @@ def _coin_balance(client, token: str) -> int:
 
 
 def test_callback_grants_coins(client) -> None:
-    """验签通过的回调(无 reward_amount → 回退 AD_REWARD_COIN)→ 发金币到账 + 计数 +1。"""
+    """验签通过且带 eCPM → 按数值公式发金币到账 + 计数 +1。"""
     phone = "13800003001"
     token = _login(client, phone)
     uid = _user_id(phone)
 
-    r = _callback(client, _signed(uid, "trans_a1", reward_name="金币"))
+    r = _callback(client, _signed(uid, "trans_a1", reward_name="金币", ecpm="200"))
     assert r.status_code == 200, r.text
     assert r.json() == {"is_verify": True, "reason": 0}
 
-    assert _coin_balance(client, token) == AD_REWARD_COIN
+    assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)
 
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 1
     assert st["daily_limit"] == DAILY_AD_REWARD_LIMIT
     assert st["remaining"] == DAILY_AD_REWARD_LIMIT - 1
-    assert st["coin_per_ad"] == AD_REWARD_COIN
+    assert st["coin_per_ad"] == 0
 
 
-def test_callback_uses_reward_amount(client) -> None:
-    """带合法 reward_amount → 按它发金币(而非常量);异常值回退/夹紧。"""
+def test_callback_without_ecpm_records_exception(client) -> None:
+    """S2S 和客户端会话都没有 eCPM → 不发金币,只记录异常状态。"""
     phone = "13800003011"
     token = _login(client, phone)
     uid = _user_id(phone)
 
-    # 合法值:发 250
-    assert _callback(client, _signed(uid, "ra_ok", reward_amount="250")).status_code == 200
-    assert _coin_balance(client, token) == 250
-
-    # ≤0 / 非数字 → 回退 AD_REWARD_COIN(累加 AD_REWARD_COIN,当前 666)
-    assert _callback(client, _signed(uid, "ra_zero", reward_amount="0")).status_code == 200
-    assert _coin_balance(client, token) == 250 + AD_REWARD_COIN
-
-    # 超上限 → 夹紧到 MAX_AD_REWARD_COIN
-    assert _callback(client, _signed(uid, "ra_big", reward_amount="999999")).status_code == 200
-    assert _coin_balance(client, token) == 250 + AD_REWARD_COIN + MAX_AD_REWARD_COIN
+    assert _callback(client, _signed(uid, "ecpm_missing")).status_code == 200
+    assert _coin_balance(client, token) == 0
+    st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
+    assert st["used_today"] == 0
 
 
 def test_callback_idempotent(client) -> None:
@@ -100,11 +91,11 @@ def test_callback_idempotent(client) -> None:
     token = _login(client, phone)
     uid = _user_id(phone)
 
-    p = _signed(uid, "trans_dup")
+    p = _signed(uid, "trans_dup", ecpm="200")
     assert _callback(client, p).status_code == 200
     assert _callback(client, p).status_code == 200  # 重试
 
-    assert _coin_balance(client, token) == AD_REWARD_COIN  # 没翻倍
+    assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)  # 没翻倍
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 1
 
@@ -134,15 +125,15 @@ def test_daily_count_cap(client, monkeypatch) -> None:
     uid = _user_id(phone)
 
     for i in range(3):
-        r = _callback(client, _signed(uid, f"trans_cap_{i}"))
+        r = _callback(client, _signed(uid, f"trans_cap_{i}", ecpm="200"))
         assert r.status_code == 200, r.text
 
     # 第 4 次:capped,仍 is_verify(不让穿山甲重试),但不加币
-    r = _callback(client, _signed(uid, "trans_cap_over"))
+    r = _callback(client, _signed(uid, "trans_cap_over", ecpm="200"))
     assert r.status_code == 200
     assert r.json() == {"is_verify": True, "reason": 0}
 
-    assert _coin_balance(client, token) == 3 * AD_REWARD_COIN
+    assert _coin_balance(client, token) == sum(calculate_ad_reward_coin("200", i) for i in range(1, 4))
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 3
     assert st["remaining"] == 0
@@ -169,7 +160,7 @@ def test_daily_watch_time_cap(client, monkeypatch) -> None:
 
     # 此时回调发奖 → 时长闸命中 → capped(is_verify 仍 true,不让重试),不加金币
     before = _coin_balance(client, token)
-    r = _callback(client, _signed(uid, "trans_time_cap"))
+    r = _callback(client, _signed(uid, "trans_time_cap", ecpm="200"))
     assert r.status_code == 200
     assert r.json() == {"is_verify": True, "reason": 0}
     assert _coin_balance(client, token) == before  # 未加币
@@ -209,7 +200,7 @@ def test_reward_status_mid_round(client) -> None:
     phone = "13800003102"
     token = _login(client, phone)
     uid = _user_id(phone)
-    _callback(client, _signed(uid, "trans_mid_1"))
+    _callback(client, _signed(uid, "trans_mid_1", ecpm="200"))
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 1
     assert st["round_count"] == 0
@@ -224,7 +215,7 @@ def test_reward_status_round_complete_enters_cooldown(client) -> None:
     token = _login(client, phone)
     uid = _user_id(phone)
     for i in range(VIDEO_ROUND_REQUIRED_COUNT):
-        _callback(client, _signed(uid, f"trans_rc_{i}"))
+        _callback(client, _signed(uid, f"trans_rc_{i}", ecpm="200"))
 
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == VIDEO_ROUND_REQUIRED_COUNT
@@ -248,7 +239,7 @@ def test_reward_status_cooldown_expired(client) -> None:
     token = _login(client, phone)
     uid = _user_id(phone)
     for i in range(VIDEO_ROUND_REQUIRED_COUNT):
-        _callback(client, _signed(uid, f"trans_exp_{i}"))
+        _callback(client, _signed(uid, f"trans_exp_{i}", ecpm="200"))
 
     # 把当日所有 granted 记录的 created_at 推到 11 分钟前(覆盖冷却末尾那条)
     db = SessionLocal()

tests/test_welfare.py

@@ -4,15 +4,20 @@
 """
 from __future__ import annotations
 
+import json
+
+from app.core.config import settings
 from app.core.rewards import (
     COIN_PER_CENT,
     COIN_PER_YUAN,
     MIN_EXCHANGE_COIN,
+    SIGNIN_BOOST_COIN,
     SIGNIN_REWARDS,
     TASK_ENABLE_NOTIFICATION,
     TASK_REWARDS,
 )
 from app.db.session import SessionLocal
+from app.integrations import pangle
 from app.repositories import wallet as crud_wallet
 from app.repositories.user import get_user_by_phone
 
@@ -29,6 +34,12 @@ def _auth(token: str) -> dict[str, str]:
     return {"Authorization": f"Bearer {token}"}
 
 
+def _signed_ad_callback(user_id: int, trans_id: str, **params: str) -> dict[str, str]:
+    body = {"user_id": str(user_id), "trans_id": trans_id, **params}
+    body["sign"] = pangle.build_sign(trans_id, settings.PANGLE_REWARD_SECRET)
+    return body
+
+
 def test_account_auto_created_empty(client) -> None:
     """首次访问账户:自动建空账户,余额全 0。"""
     token = _login(client, "13800001001")
@@ -91,8 +102,9 @@ def test_signin_flow(client) -> None:
 
 
 def test_signin_boost_flow(client) -> None:
-    """签到后看广告膨胀 → 补发一笔等额签到金币,每天只能膨胀一次。"""
-    token = _login(client, "13800001011")
+    """签到后看广告膨胀 → S2S 固定补发 2000 金币,每天只能膨胀一次。"""
+    phone = "13800001011"
+    token = _login(client, phone)
 
     r = client.post("/api/v1/signin/boost", json={}, headers=_auth(token))
     assert r.status_code == 409
@@ -101,14 +113,30 @@ def test_signin_boost_flow(client) -> None:
     assert r.status_code == 200, r.text
     first_coin = r.json()["coin_awarded"]
 
-    r = client.post("/api/v1/signin/boost", json={"ad_ref_id": "ad-session-1"}, headers=_auth(token))
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        uid = user.id
+
+    extra = json.dumps({"reward_scene": "signin_boost", "ad_session_id": "signin-session-1"})
+    r = client.get(
+        "/api/v1/ad/pangle-callback",
+        params=_signed_ad_callback(uid, "signin-boost-trans-1", extra=extra, ecpm="200"),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json() == {"is_verify": True, "reason": 0}
+
+    r = client.post("/api/v1/signin/boost", json={"ad_ref_id": "signin-boost-trans-1"}, headers=_auth(token))
     assert r.status_code == 200, r.text
     body = r.json()
-    assert body["coin_awarded"] == first_coin
-    assert body["coin_balance"] == first_coin * 2
+    assert body["coin_awarded"] == SIGNIN_BOOST_COIN
+    assert body["coin_balance"] == first_coin + SIGNIN_BOOST_COIN
 
-    r = client.post("/api/v1/signin/boost", json={"ad_ref_id": "ad-session-1"}, headers=_auth(token))
-    assert r.status_code == 409
+    r = client.get(
+        "/api/v1/ad/pangle-callback",
+        params=_signed_ad_callback(uid, "signin-boost-trans-2", extra=extra, ecpm="200"),
+    )
+    assert r.status_code == 200, r.text
 
     r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
     types = [t["biz_type"] for t in r.json()["items"]]