Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/analytics-success-rate

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

alembic/versions/11c44afbea58_analytics_selfstat_tables.py

@@ -0,0 +1,72 @@
+"""analytics_selfstat tables
+
+Revision ID: 11c44afbea58
+Revises: admin_user_plain_password
+Create Date: 2026-07-08 16:32:49.351817
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = '11c44afbea58'
+down_revision: str | Sequence[str] | None = 'admin_user_plain_password'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'analytics_selfstat',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('device_id', sa.String(length=64), nullable=False),
+        sa.Column('epoch_id', sa.String(length=64), nullable=False),
+        sa.Column('app_ver', sa.String(length=32), nullable=True),
+        sa.Column('oem', sa.String(length=32), nullable=True),
+        sa.Column('os', sa.String(length=32), nullable=True),
+        sa.Column('batches_attempted', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('batches_ok', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('batches_fail', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('retries', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('queue_depth', sa.Integer(), nullable=False, server_default='0'),
+        sa.Column('sent_at', sa.BigInteger(), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True),
+                  server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_analytics_selfstat_created_at'), ['created_at'], unique=False)
+        batch_op.create_index(batch_op.f('ix_analytics_selfstat_device_id'), ['device_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_analytics_selfstat_epoch_id'), ['epoch_id'], unique=False)
+
+    op.create_table(
+        'analytics_selfstat_event',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('snapshot_id', sa.Integer(), nullable=False),
+        sa.Column('event', sa.String(length=64), nullable=False),
+        sa.Column('attempted', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('drop_capture', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('delivered', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.Column('drop_undelivered', sa.BigInteger(), nullable=False, server_default='0'),
+        sa.ForeignKeyConstraint(['snapshot_id'], ['analytics_selfstat.id'], ),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_event'), ['event'], unique=False)
+        batch_op.create_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'), ['snapshot_id'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('analytics_selfstat_event', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_snapshot_id'))
+        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_event_event'))
+    op.drop_table('analytics_selfstat_event')
+    with op.batch_alter_table('analytics_selfstat', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_epoch_id'))
+        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_device_id'))
+        batch_op.drop_index(batch_op.f('ix_analytics_selfstat_created_at'))
+    op.drop_table('analytics_selfstat')

app/admin/main.py

@@ -26,6 +26,7 @@ from app.admin.routers.cps import router as cps_router
 from app.admin.routers.dashboard import router as dashboard_router
 from app.admin.routers.device_liveness import router as device_liveness_router
 from app.admin.routers.ops_stat_config import router as ops_stat_config_router
+from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.event_logs import router as event_logs_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
@@ -97,6 +98,7 @@ admin_app.include_router(withdraw_router)
 admin_app.include_router(price_report_router)
 admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
+admin_app.include_router(analytics_health_router)
 admin_app.include_router(feedback_qr_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(roles_router)

app/admin/repositories/analytics_health.py

@@ -0,0 +1,146 @@
+"""埋点健康度聚合(埋点成功率 / 上报成功率)。
+
+只存原始累计快照,查询时在 Python 侧差分聚合(admin 低频、量级小,跨 PG/SQLite 无方言坑;
+与 cps.py / coupon_data.py 同款约定)。差分按 (device_id, epoch_id, event) 分区、created_at
+升序,相邻做差、负值夹 0;每增量按其快照 created_at 归入北京天桶。
+"""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import UTC, datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core import rewards
+from app.models.analytics_selfstat import AnalyticsSelfStat as H
+from app.models.analytics_selfstat import AnalyticsSelfStatEvent as E
+
+_COUNTS = ("attempted", "drop_capture", "delivered", "drop_undelivered")
+
+
+def diff_snapshots(rows: list[dict]) -> list[dict]:
+    """累计快照行 → 每快照增量行(纯逻辑)。
+
+    rows 每行含 device_id/epoch_id/event/created_at/app_ver/oem/os + 四个累计计数。
+    返回每行含 dims + created_at + 四个增量 d_*(分区首行增量=累计值;负值夹 0)。
+    """
+    parts: dict[tuple, list[dict]] = defaultdict(list)
+    for r in rows:
+        parts[(r["device_id"], r["epoch_id"], r["event"])].append(r)
+
+    out: list[dict] = []
+    for group in parts.values():
+        group.sort(key=lambda r: (r["created_at"], r.get("id", 0)))
+        prev = {k: 0 for k in _COUNTS}
+        for r in group:
+            deltas = {f"d_{k}": max(0, int(r[k]) - prev[k]) for k in _COUNTS}
+            out.append({
+                "device_id": r["device_id"], "epoch_id": r["epoch_id"], "event": r["event"],
+                "created_at": r["created_at"], "app_ver": r["app_ver"],
+                "oem": r["oem"], "os": r["os"], **deltas,
+            })
+            prev = {k: int(r[k]) for k in _COUNTS}
+    return out
+
+
+def _cn_day(dt: datetime) -> str:
+    """created_at(UTC 口径)→ 北京日期字符串 YYYY-MM-DD。naive 当 UTC,tz-aware 直接换算。"""
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return dt.astimezone(rewards.CN_TZ).date().isoformat()
+
+
+def _rates(sums: dict) -> dict:
+    """由四个增量和派生两段率(分母 0 → None)。"""
+    persisted_denom = sums["attempted"]
+    report_denom = sums["delivered"] + sums["drop_undelivered"]
+    return {
+        **sums,
+        "track_success_rate": (
+            (sums["attempted"] - sums["drop_capture"]) / persisted_denom
+            if persisted_denom else None
+        ),
+        "report_success_rate": (
+            sums["delivered"] / report_denom if report_denom else None
+        ),
+    }
+
+
+def _sum_deltas(deltas: list[dict]) -> dict:
+    return {k: sum(d[f"d_{k}"] for d in deltas) for k in _COUNTS}
+
+
+def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
+    """取 [from, to) 区间行 + 每分区在 from 左侧的最后一条基线行(供第一条区间增量做差)。"""
+    cols = (
+        H.id, H.device_id, H.epoch_id, E.event, H.created_at,
+        H.app_ver, H.oem, H.os,
+        E.attempted, E.drop_capture, E.delivered, E.drop_undelivered,
+    )
+    in_range = db.execute(
+        select(*cols).join(E, E.snapshot_id == H.id)
+        .where(H.created_at >= date_from, H.created_at < date_to)
+    ).mappings().all()
+
+    # 注:基线子查询无下界扫 from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。
+    # 用 max(id) 而非 max(created_at) 选"最新一条":id 严格单调,避免 SQLite 秒级时间戳撞车时选歧义。
+    sub = (
+        select(H.device_id, H.epoch_id, E.event, func.max(H.id).label("max_id"))
+        .join(E, E.snapshot_id == H.id)
+        .where(H.created_at < date_from)
+        .group_by(H.device_id, H.epoch_id, E.event)
+        .subquery()
+    )
+    baseline = db.execute(
+        select(*cols).join(E, E.snapshot_id == H.id).join(
+            sub, sub.c.max_id == H.id
+        )
+    ).mappings().all()
+
+    return [dict(r) for r in list(baseline) + list(in_range)]
+
+
+def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
+    """差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。
+
+    Python 侧过滤需对齐 tz 口径:SQLite 返回 naive UTC,PG 返回 aware UTC。
+    统一转成 naive UTC 再比较,兼容两种后端。
+    """
+    def _to_naive_utc(dt: datetime) -> datetime:
+        if dt.tzinfo is not None:
+            return dt.astimezone(UTC).replace(tzinfo=None)
+        return dt
+
+    from_naive = _to_naive_utc(date_from)
+    to_naive = _to_naive_utc(date_to)
+    deltas = diff_snapshots(_fetch_rows(db, date_from, date_to))
+    return [d for d in deltas if from_naive <= _to_naive_utc(d["created_at"]) < to_naive]
+
+
+def overview(db: Session, date_from: datetime, date_to: datetime) -> dict:
+    deltas = _in_range_deltas(db, date_from, date_to)
+    return _rates(_sum_deltas(deltas))
+
+
+def trend(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
+    deltas = _in_range_deltas(db, date_from, date_to)
+    by_day: dict[str, list[dict]] = defaultdict(list)
+    for d in deltas:
+        by_day[_cn_day(d["created_at"])].append(d)
+    return [
+        {"day": day, **_rates(_sum_deltas(items))}
+        for day, items in sorted(by_day.items())
+    ]
+
+
+def breakdown(db: Session, date_from: datetime, date_to: datetime, dim: str) -> list[dict]:
+    if dim not in ("event", "app_ver", "oem"):
+        raise ValueError(f"invalid dim: {dim!r}")
+    deltas = _in_range_deltas(db, date_from, date_to)
+    by_key: dict[str, list[dict]] = defaultdict(list)
+    for d in deltas:
+        by_key[d.get(dim) or "(unknown)"].append(d)
+    rows = [{"key": key, **_rates(_sum_deltas(items))} for key, items in by_key.items()]
+    rows.sort(key=lambda r: (r["report_success_rate"] is None, r["report_success_rate"] or 0.0))
+    return rows

app/admin/routers/analytics_health.py

@@ -0,0 +1,49 @@
+"""admin 埋点健康度:埋点成功率 / 上报成功率 总览 + 趋势 + 下钻(只读)。"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query
+
+from app.admin.deps import AdminDb, get_current_admin
+from app.admin.repositories import analytics_health as repo
+from app.admin.schemas.analytics_health import (
+    HealthBreakdownRow,
+    HealthMetrics,
+    HealthTrendPoint,
+)
+
+router = APIRouter(
+    prefix="/admin/api/analytics-health",
+    tags=["admin-analytics-health"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+
+@router.get("/overview", response_model=HealthMetrics, summary="两段成功率总览")
+def overview(
+    db: AdminDb,
+    date_from: Annotated[datetime, Query()],
+    date_to: Annotated[datetime, Query()],
+) -> HealthMetrics:
+    return HealthMetrics(**repo.overview(db, date_from, date_to))
+
+
+@router.get("/trend", response_model=list[HealthTrendPoint], summary="按北京天趋势")
+def trend(
+    db: AdminDb,
+    date_from: Annotated[datetime, Query()],
+    date_to: Annotated[datetime, Query()],
+) -> list[HealthTrendPoint]:
+    return [HealthTrendPoint(**p) for p in repo.trend(db, date_from, date_to)]
+
+
+@router.get("/breakdown", response_model=list[HealthBreakdownRow], summary="按维度下钻")
+def breakdown(
+    db: AdminDb,
+    date_from: Annotated[datetime, Query()],
+    date_to: Annotated[datetime, Query()],
+    dim: Annotated[str, Query(pattern="^(event|app_ver|oem)$")] = "event",
+) -> list[HealthBreakdownRow]:
+    return [HealthBreakdownRow(**r) for r in repo.breakdown(db, date_from, date_to, dim)]

app/admin/schemas/analytics_health.py

@@ -0,0 +1,21 @@
+"""埋点健康度 admin 响应 schema。"""
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class HealthMetrics(BaseModel):
+    attempted: int
+    drop_capture: int
+    delivered: int
+    drop_undelivered: int
+    track_success_rate: float | None
+    report_success_rate: float | None
+
+
+class HealthTrendPoint(HealthMetrics):
+    day: str
+
+
+class HealthBreakdownRow(HealthMetrics):
+    key: str

app/api/v1/analytics.py

@@ -6,13 +6,18 @@ POST /api/v1/analytics/events — 批量接收新手引导(及后续)埋点,appe
 """
 from __future__ import annotations
 
-from fastapi import APIRouter, Request
+import logging
+
+from fastapi import APIRouter, HTTPException, Request
 
 from app.api.deps import DbSession
 from app.repositories import analytics as analytics_repo
+from app.repositories import analytics_selfstat as selfstat_repo
 from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut
+from app.schemas.analytics_selfstat import SelfStatBatchIn, SelfStatIngestOut
 
 router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"])
+logger = logging.getLogger("shagua.analytics")
 
 
 def _client_ip(request: Request) -> str:
@@ -29,3 +34,13 @@ def ingest_events(
 ) -> AnalyticsIngestOut:
     n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request))
     return AnalyticsIngestOut(received=n)
+
+
+@router.post("/selfstat", response_model=SelfStatIngestOut, summary="上报自报计数快照")
+def ingest_selfstat(batch: SelfStatBatchIn, db: DbSession) -> SelfStatIngestOut:
+    try:
+        snap_id = selfstat_repo.record_selfstat(db, batch)
+    except Exception:  # noqa: BLE001 — 计数链路要稳,落库失败不裸奔 500,记日志回明确错误
+        logger.exception("selfstat ingest failed device=%s epoch=%s", batch.device_id, batch.epoch_id)
+        raise HTTPException(status_code=503, detail="selfstat ingest failed") from None
+    return SelfStatIngestOut(snapshot_id=snap_id)

app/api/v1/wallet.py

@@ -43,7 +43,6 @@ from app.schemas.welfare import (
     WithdrawRequest,
     WithdrawResultOut,
     WithdrawStatusOut,
-    WithdrawTierOut,
 )
 
 logger = logging.getLogger("shagua.wallet")
@@ -174,15 +173,8 @@ def unbind_wechat(
     return UnbindWechatResultOut(bound=False)
 
 
-@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态/免确认开关/档位")
-def withdraw_info(
-    user: CurrentUser,
-    db: DbSession,
-    source: str = Query(
-        "coin_cash",
-        description="提现账户:coin_cash(福利页,下发 tiers 档位) / invite_cash(邀请页,tiers 为空走旧逻辑)",
-    ),
-) -> WithdrawInfoOut:
+@router.get("/withdraw-info", response_model=WithdrawInfoOut, summary="提现额度/绑定状态/免确认开关")
+def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut:
     u = db.get(User, user.id)
     # 顺带同步免确认授权状态(捕获首单确认后已生效的授权 pending→active),让开关展示实时
     auth = crud_wallet.sync_transfer_auth(db, user.id)
@@ -193,7 +185,6 @@ def withdraw_info(
         wechat_nickname=u.wechat_nickname if u else None,
         wechat_avatar_url=u.wechat_avatar_url if u else None,
         transfer_auth_enabled=bool(auth and auth.state == "active"),
-        tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
     )
 
 
@@ -227,9 +218,6 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
             status_code=status.HTTP_409_CONFLICT,
             detail="已有提现申请正在审核或打款中,请处理完成后再申请",
         ) from e
-    except crud_wallet.WithdrawTierUnavailableError as e:
-        # 福利页档位闸(7-9):次数满/已选其他额度。正常客户端已按 tiers 预拦,此处兜底防绕过。
-        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="现金余额不足") from e
 

app/core/rewards.py

@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 from datetime import date, datetime, timedelta, timezone
-from typing import NamedTuple
 
 # 业务时区:签到的"今天"按北京时间算,不能用 UTC。
 # 否则 UTC+8 的凌晨 0~8 点会被算成 UTC 的前一天,导致签到日期错位。
@@ -53,30 +52,6 @@ WITHDRAW_MIN_CENTS: int = 10
 WITHDRAW_MAX_CENTS: int = 5_000_000  # 5 万元
 
 
-# ===== 提现档位(福利页 coin_cash;7-9 对齐原型 withdrawal.html)=====
-# 后端是档位唯一真相源:withdraw-info 按此下发,create_withdraw 按此校验(防绕过客户端刷)。
-# 规则(2026-07-09 拍板):
-#   - 新人档(is_newbie):账号历史一次性,"发起就算用过"(任意状态含被拒),用过即不再下发;
-#     0.1 与 0.3 各自独立同天可各提一次,且不参与常规档"每日选一个额度"互斥。
-#   - 常规档:按北京日计次(0.5×3 / 10×1 / 20×1),三档每天只能选一个。
-# invite_cash(邀请页)本轮无档位概念,不在此表。改档位=改这里发版。
-class WithdrawTier(NamedTuple):
-    amount_cents: int
-    label: str          # 客户端档位方块展示文案
-    badge: str | None   # 角标文案;None=无角标
-    daily_limit: int    # 每日次数上限(新人档的"历史一次性"另由 is_newbie 判定)
-    is_newbie: bool
-
-
-WITHDRAW_TIERS_COIN_CASH: tuple[WithdrawTier, ...] = (
-    WithdrawTier(10, "0.1", "新人福利", 1, True),
-    WithdrawTier(30, "0.3", "新人福利", 1, True),
-    WithdrawTier(50, "0.5", None, 3, False),
-    WithdrawTier(1000, "10", None, 1, False),
-    WithdrawTier(2000, "20", None, 1, False),
-)
-
-
 # ===== 一次性任务(领一次,user_task 去重)=====
 TASK_ENABLE_NOTIFICATION = "enable_notification"
 

app/models/__init__.py

@@ -7,6 +7,10 @@ from app.models.ad_watch_log import AdWatchLog  # noqa: F401
 from app.models.admin import AdminAuditLog, AdminUser  # noqa: F401
 from app.models.admin_role import AdminRole  # noqa: F401
 from app.models.analytics_event import AnalyticsEvent  # noqa: F401
+from app.models.analytics_selfstat import (  # noqa: F401
+    AnalyticsSelfStat,
+    AnalyticsSelfStatEvent,
+)
 from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.cps_activity import CpsActivity  # noqa: F401

app/models/analytics_selfstat.py

@@ -0,0 +1,59 @@
+"""埋点/上报成功率自报计数快照表(append-only)。
+
+客户端周期上报「自 epoch 起算的累计计数」;服务端只存原始快照,查询时在 Python 侧差分聚合
+(见 app/admin/repositories/analytics_health.py)。与既有 analytics_event 表完全独立。
+
+- analytics_selfstat        :一快照一行(快照头 + 设备维度 + 设备级诊断量)
+- analytics_selfstat_event  :一 event 一行(四类累计计数),外键指向快照头
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class AnalyticsSelfStat(Base):
+    __tablename__ = "analytics_selfstat"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    device_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    epoch_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    # 设备维度(每设备固定,下钻用)
+    app_ver: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    oem: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    os: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # 设备级诊断量(累计;queue_depth 是瞬时 gauge)
+    batches_attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    batches_ok: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    batches_fail: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    retries: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    queue_depth: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    sent_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True)  # 端上报时刻 epoch ms
+    # 服务端接收时间(权威,用于时间分桶与分区排序)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"<AnalyticsSelfStat id={self.id} device={self.device_id} epoch={self.epoch_id}>"
+
+
+class AnalyticsSelfStatEvent(Base):
+    __tablename__ = "analytics_selfstat_event"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    snapshot_id: Mapped[int] = mapped_column(
+        ForeignKey("analytics_selfstat.id"), index=True, nullable=False
+    )
+    event: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    drop_capture: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    delivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    drop_undelivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"<AnalyticsSelfStatEvent snap={self.snapshot_id} event={self.event}>"

app/repositories/analytics_selfstat.py

@@ -0,0 +1,38 @@
+"""自报计数快照落库。一次事务:插 1 条快照头 + N 条 event 行,返回快照 id。"""
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from app.models.analytics_selfstat import AnalyticsSelfStat, AnalyticsSelfStatEvent
+from app.schemas.analytics_selfstat import SelfStatBatchIn
+
+
+def record_selfstat(db: Session, batch: SelfStatBatchIn) -> int:
+    snap = AnalyticsSelfStat(
+        device_id=batch.device_id,
+        epoch_id=batch.epoch_id,
+        app_ver=batch.app_ver,
+        oem=batch.oem,
+        os=batch.os,
+        batches_attempted=batch.batches_attempted,
+        batches_ok=batch.batches_ok,
+        batches_fail=batch.batches_fail,
+        retries=batch.retries,
+        queue_depth=batch.queue_depth,
+        sent_at=batch.sent_at,
+    )
+    db.add(snap)
+    db.flush()  # 拿到 snap.id
+    db.add_all([
+        AnalyticsSelfStatEvent(
+            snapshot_id=snap.id,
+            event=e.event,
+            attempted=e.attempted,
+            drop_capture=e.drop_capture,
+            delivered=e.delivered,
+            drop_undelivered=e.drop_undelivered,
+        )
+        for e in batch.events
+    ])
+    db.commit()
+    return snap.id

app/repositories/wallet.py

@@ -11,7 +11,7 @@ import unicodedata
 import uuid
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import func, select, update
+from sqlalchemy import select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -67,10 +67,6 @@ class WithdrawTooFrequentError(Exception):
     """提现申请过于频繁,或已有未完成提现单。"""
 
 
-class WithdrawTierUnavailableError(Exception):
-    """该档位今日不可提:次数已满,或今天已选了其他额度(7-9 福利页档位规则)。"""
-
-
 class WithdrawTransferError(Exception):
     """调用微信转账失败(已退回余额)。"""
 
@@ -609,89 +605,6 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
         db.commit()
 
 
-def _beijing_today_start_utc() -> datetime:
-    """北京时今日 0 点(转 UTC)。WithdrawOrder.created_at 是 func.now()(UTC)存储,
-    比较时统一转 UTC,与 admin 看板 today_start 同口径(admin/repositories/queries.py)。"""
-    return (
-        datetime.now(rewards.CN_TZ)
-        .replace(hour=0, minute=0, second=0, microsecond=0)
-        .astimezone(timezone.utc)
-    )
-
-
-def withdraw_tier_states(db: Session, user_id: int, source: str = "coin_cash") -> list[dict]:
-    """福利页(coin_cash)提现档位的可提现状态。withdraw-info 下发与 create_withdraw 校验共用此口径。
-
-    规则(2026-07-09 拍板,7-9提现ui对齐):
-      - 新人档(0.1/0.3):账号历史一次性——只要发起过(**任意状态**,含被拒/失败,"发起就算")
-        即视为已用,直接**从返回列表消失**;两档各自独立互不影响,不参与"每日选一个额度"互斥。
-      - 常规档(0.5×3 / 10×1 / 20×1):按北京日计次,"发起就算占用"(当天创建的单不论最终状态
-        都计入,被拒/失败不退当天名额);三档每天只能选一个,选定后其余两档当天 other_tier_selected。
-      - invite_cash 本轮无档位概念 → 返回空列表(邀请页客户端仍用本地写死档位,行为不变)。
-    余额是否足够由客户端本地判断(余额随兑换实时变化,不在此快照)。
-    """
-    if source != "coin_cash":
-        return []
-    tiers = rewards.WITHDRAW_TIERS_COIN_CASH
-    amounts = [t.amount_cents for t in tiers]
-    newbie_amounts = [t.amount_cents for t in tiers if t.is_newbie]
-    # 新人档历史是否用过:任意时间、任意状态("发起就算")
-    used_newbie: set[int] = set(
-        db.execute(
-            select(WithdrawOrder.amount_cents)
-            .distinct()
-            .where(
-                WithdrawOrder.user_id == user_id,
-                WithdrawOrder.source == "coin_cash",
-                WithdrawOrder.amount_cents.in_(newbie_amounts),
-            )
-        ).scalars()
-    ) if newbie_amounts else set()
-    # 今日(北京日)每档已发起次数(任意状态)
-    today_counts: dict[int, int] = {
-        int(amount): int(cnt)
-        for amount, cnt in db.execute(
-            select(WithdrawOrder.amount_cents, func.count(WithdrawOrder.id))
-            .where(
-                WithdrawOrder.user_id == user_id,
-                WithdrawOrder.source == "coin_cash",
-                WithdrawOrder.amount_cents.in_(amounts),
-                WithdrawOrder.created_at >= _beijing_today_start_utc(),
-            )
-            .group_by(WithdrawOrder.amount_cents)
-        )
-    }
-    # "每日选一个额度":今天发起过的常规档(新人档不算)
-    selected_regular = next(
-        (t.amount_cents for t in tiers if not t.is_newbie and today_counts.get(t.amount_cents, 0) > 0),
-        None,
-    )
-    out: list[dict] = []
-    for t in tiers:
-        if t.is_newbie:
-            if t.amount_cents in used_newbie:
-                continue  # 用过即消失,不再下发
-            out.append({
-                "amount_cents": t.amount_cents, "label": t.label, "badge": t.badge,
-                "is_newbie": True, "available": True, "disabled_reason": None,
-                "remaining_today": 1,
-            })
-            continue
-        used = today_counts.get(t.amount_cents, 0)
-        if selected_regular is not None and selected_regular != t.amount_cents:
-            available, reason, remaining = False, "other_tier_selected", 0
-        elif used >= t.daily_limit:
-            available, reason, remaining = False, "quota_exhausted", 0
-        else:
-            available, reason, remaining = True, None, t.daily_limit - used
-        out.append({
-            "amount_cents": t.amount_cents, "label": t.label, "badge": t.badge,
-            "is_newbie": False, "available": available, "disabled_reason": reason,
-            "remaining_today": remaining,
-        })
-    return out
-
-
 def create_withdraw(
     db: Session,
     user_id: int,
@@ -747,19 +660,6 @@ def create_withdraw(
     if active_order_id is not None:
         raise WithdrawTooFrequentError
 
-    # 福利页档位闸(7-9):coin_cash 只能提预设档位,且该档今日可提(服务端权威口径,防绕过
-    # 客户端刷)。放在幂等返回/在途互斥之后:同号重试仍原样返回旧单,不被档位闸误杀。
-    # allow_sub_min(0.01 调试直发)保持原样放行,不受档位约束;invite_cash 本轮无档位概念不校验。
-    if source == "coin_cash" and not allow_sub_min:
-        tier_state = next(
-            (t for t in withdraw_tier_states(db, user_id, source) if t["amount_cents"] == amount_cents),
-            None,
-        )
-        if tier_state is None:  # 非预设档位金额,或新人档已用过(已从列表消失)
-            raise InvalidWithdrawAmountError
-        if not tier_state["available"]:
-            raise WithdrawTierUnavailableError
-
     # 账户须存在(原子扣款的 UPDATE 不会建账户)
     get_or_create_account(db, user_id, commit=True)
 

app/schemas/analytics_selfstat.py

@@ -0,0 +1,34 @@
+"""自报计数上报 schema。字段名对齐客户端 payload(snake_case),累计值语义。"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class SelfStatEventIn(BaseModel):
+    event: str = Field(max_length=64)
+    attempted: int = 0
+    drop_capture: int = 0
+    delivered: int = 0
+    drop_undelivered: int = 0
+
+
+class SelfStatBatchIn(BaseModel):
+    device_id: str = Field(max_length=64)
+    epoch_id: str = Field(max_length=64)
+    sent_at: int | None = None
+    app_ver: str | None = Field(default=None, max_length=32)
+    oem: str | None = Field(default=None, max_length=32)
+    os: str | None = Field(default=None, max_length=32)
+    batches_attempted: int = 0
+    batches_ok: int = 0
+    batches_fail: int = 0
+    retries: int = 0
+    queue_depth: int = 0
+    # 允许空列表:某次快照只上报设备级计数(batches_*/retries/queue_depth)、无 event 细分时也合法
+    # (与 AnalyticsBatchIn 的 min_length=1 有意不同——那是行为事件、必须至少一条)。
+    events: list[SelfStatEventIn] = Field(default_factory=list, max_length=200)
+
+
+class SelfStatIngestOut(BaseModel):
+    ok: bool = True
+    snapshot_id: int

app/schemas/welfare.py

@@ -75,21 +75,6 @@ class ExchangeResultOut(BaseModel):
 
 # ===== 提现(现金 → 微信零钱) =====
 
-class WithdrawTierOut(BaseModel):
-    """提现档位(福利页 coin_cash;7-9 对齐原型)。served by rewards.WITHDRAW_TIERS_COIN_CASH。"""
-
-    amount_cents: int = Field(..., description="档位金额(分)")
-    label: str = Field(..., description="档位方块展示文案,如 0.1 / 10")
-    badge: str | None = Field(None, description="角标文案(如 新人福利);无则空")
-    is_newbie: bool = Field(False, description="新人档:历史一次性,用过后不再下发;免广告直提")
-    available: bool = Field(True, description="当前是否可提(次数/选一额度口径;余额由客户端自判)")
-    disabled_reason: str | None = Field(
-        None,
-        description="不可提原因:quota_exhausted(今日次数满) / other_tier_selected(今日已选其他额度)",
-    )
-    remaining_today: int = Field(0, description="今日剩余可提次数")
-
-
 class WithdrawInfoOut(BaseModel):
     min_cents: int = Field(..., description="单次最低提现(分)")
     max_cents: int = Field(..., description="单次最高提现(分)")
@@ -99,10 +84,6 @@ class WithdrawInfoOut(BaseModel):
     transfer_auth_enabled: bool = Field(
         False, description="是否已开启免确认到账(开启后提现免跳微信确认,直接到账)"
     )
-    tiers: list[WithdrawTierOut] = Field(
-        default_factory=list,
-        description="提现档位(source=coin_cash 下发;invite_cash 为空,客户端走旧逻辑)",
-    )
 
 
 # ===== 免确认收款授权(用户授权免确认模式)=====

tests/test_analytics_health.py

@@ -0,0 +1,160 @@
+"""埋点健康度聚合:纯差分函数 + admin 端点。"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from app.admin.repositories.analytics_health import _cn_day, _rates, diff_snapshots
+
+
+def _row(device, epoch, event, ts, **cum) -> dict:
+    base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}
+    base.update(cum)
+    return {
+        "device_id": device, "epoch_id": epoch, "event": event,
+        "created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc),
+        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+        **base,
+    }
+
+
+def test_diff_first_row_is_full_cumulative() -> None:
+    rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)]
+    out = diff_snapshots(rows)
+    assert len(out) == 1
+    assert out[0]["d_attempted"] == 100
+    assert out[0]["d_delivered"] == 90
+
+
+def test_diff_consecutive_delta() -> None:
+    rows = [
+        _row("d", "e", "video_play", 1, attempted=100, delivered=90),
+        _row("d", "e", "video_play", 2, attempted=150, delivered=140),
+    ]
+    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
+    assert out[1]["d_attempted"] == 50
+    assert out[1]["d_delivered"] == 50
+
+
+def test_diff_epoch_reset_new_partition() -> None:
+    rows = [
+        _row("d", "e1", "video_play", 1, attempted=100),
+        _row("d", "e2", "video_play", 2, attempted=5),
+    ]
+    out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)}
+    assert out["e1"] == 100
+    assert out["e2"] == 5
+
+
+def test_diff_clamps_negative_on_reorder() -> None:
+    rows = [
+        _row("d", "e", "video_play", 1, attempted=100),
+        _row("d", "e", "video_play", 2, attempted=80),
+    ]
+    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
+    assert out[1]["d_attempted"] == 0
+
+
+def test_rates_computes_both_formulas() -> None:
+    out = _rates({"attempted": 150, "drop_capture": 0, "delivered": 140, "drop_undelivered": 10})
+    assert out["track_success_rate"] == 1.0
+    assert out["report_success_rate"] == 140 / 150
+
+
+def test_rates_zero_denominator_yields_none() -> None:
+    out = _rates({"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0})
+    assert out["track_success_rate"] is None
+    assert out["report_success_rate"] is None
+
+
+def test_cn_day_beijing_boundary() -> None:
+    # UTC 15:59 → 北京 23:59 同日;UTC 16:00 → 北京 次日 00:00
+    assert _cn_day(datetime(2026, 7, 1, 15, 59, tzinfo=timezone.utc)) == "2026-07-01"
+    assert _cn_day(datetime(2026, 7, 1, 16, 0, tzinfo=timezone.utc)) == "2026-07-02"
+
+
+def test_diff_same_timestamp_ordered_by_id() -> None:
+    ts = datetime(2026, 7, 1, 1, 0, tzinfo=timezone.utc)
+    # 相同 created_at,乱序传入(id=2 在前);应按 id 排序 → id=1(cum100) 在前
+    rows = [
+        {"id": 2, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
+         "app_ver": "v", "oem": "o", "os": "s",
+         "attempted": 150, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
+        {"id": 1, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
+         "app_ver": "v", "oem": "o", "os": "s",
+         "attempted": 100, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
+    ]
+    out = diff_snapshots(rows)
+    assert out[0]["d_attempted"] == 100  # id=1 sorts first → its cumulative
+    assert out[1]["d_attempted"] == 50   # id=2 second → 150-100
+
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.admin.main import admin_app
+from app.admin.repositories import admin_user as admin_repo
+from app.db.session import SessionLocal
+
+
+@pytest.fixture()
+def admin_client() -> TestClient:
+    return TestClient(admin_app)
+
+
+@pytest.fixture()
+def admin_token() -> str:
+    db = SessionLocal()
+    try:
+        if admin_repo.get_by_username(db, "health_admin") is None:
+            admin_repo.create_admin(db, username="health_admin", password="pw", role="super_admin")
+    finally:
+        db.close()
+    c = TestClient(admin_app)
+    r = c.post("/admin/api/auth/login", json={"username": "health_admin", "password": "pw"})
+    return r.json()["access_token"]
+
+
+def _auth(t: str) -> dict:
+    return {"Authorization": f"Bearer {t}"}
+
+
+def _seed_two_snapshots(client: TestClient) -> None:
+    for attempted, delivered in ((100, 90), (150, 140)):
+        client.post("/api/v1/analytics/selfstat", json={
+            "device_id": "d-health", "epoch_id": "e-health",
+            "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+            "events": [{"event": "video_play", "attempted": attempted,
+                        "drop_capture": 0, "delivered": delivered, "drop_undelivered": 0}],
+        })
+
+
+def test_health_overview_requires_auth(admin_client: TestClient) -> None:
+    r = admin_client.get("/admin/api/analytics-health/overview",
+                         params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"})
+    assert r.status_code == 401
+
+
+def test_health_overview(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
+    _seed_two_snapshots(client)  # 播种走公开 app(ingest 端点在 app,不在 admin_app)
+    r = admin_client.get(
+        "/admin/api/analytics-health/overview",
+        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    data = r.json()
+    # 两快照累计 100→150,差分后 attempted 总 150(首快照 100 + 增量 50)
+    assert data["attempted"] == 150
+    assert data["track_success_rate"] == 1.0
+
+
+def test_health_breakdown(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
+    _seed_two_snapshots(client)  # 播种走公开 app
+    r = admin_client.get(
+        "/admin/api/analytics-health/breakdown",
+        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z", "dim": "event"},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    rows = r.json()
+    assert any(row["key"] == "video_play" for row in rows)

tests/test_analytics_selfstat.py

@@ -0,0 +1,43 @@
+"""自报计数上报端点测试。"""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+
+def _payload(**over) -> dict:
+    base = {
+        "device_id": "dev-1", "epoch_id": "ep-1", "sent_at": 1700000000000,
+        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+        "batches_attempted": 10, "batches_ok": 9, "batches_fail": 1,
+        "retries": 1, "queue_depth": 2,
+        "events": [
+            {"event": "video_play", "attempted": 100, "drop_capture": 1,
+             "delivered": 95, "drop_undelivered": 2},
+        ],
+    }
+    base.update(over)
+    return base
+
+
+def test_selfstat_ingest_ok(client: TestClient) -> None:
+    r = client.post("/api/v1/analytics/selfstat", json=_payload())
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["ok"] is True
+    assert isinstance(body["snapshot_id"], int)
+
+
+def test_selfstat_ingest_empty_events(client: TestClient) -> None:
+    r = client.post("/api/v1/analytics/selfstat", json=_payload(events=[]))
+    assert r.status_code == 200, r.text
+    assert r.json()["ok"] is True
+
+
+def test_selfstat_ingest_db_error_returns_503(client: TestClient, monkeypatch) -> None:
+    """落库异常时端点回 503(计数链路稳定性守卫),不裸奔 500。"""
+    def _boom(*_args, **_kwargs):
+        raise RuntimeError("db gone")
+
+    monkeypatch.setattr("app.repositories.analytics_selfstat.record_selfstat", _boom)
+    r = client.post("/api/v1/analytics/selfstat", json=_payload())
+    assert r.status_code == 503, r.text

tests/test_invite_cash_withdraw.py

@@ -143,15 +143,14 @@ def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     _reject(r1.json()["out_bill_no"])    # 退回 invite_cash + 结清活跃单
     r2 = client.post(
         "/api/v1/wallet/withdraw",
-        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
-        json={"amount_cents": 50, "source": "coin_cash"},
+        json={"amount_cents": 100, "source": "coin_cash"},
         headers=_auth(token),
     )
     assert r2.json()["status"] == "reviewing"
 
     cash, invite_cash = _balances(client, token)
     assert invite_cash == 500    # 已退回
-    assert cash == 350           # 扣了 cash 50
+    assert cash == 300           # 扣了 cash 100
 
 
 def test_invite_me_returns_reward_stats(client) -> None:
@@ -178,8 +177,7 @@ def test_withdraw_orders_source_filter(client, monkeypatch) -> None:
     _reject(r1.json()["out_bill_no"])    # 结清,才能提第二笔
     client.post(
         "/api/v1/wallet/withdraw",
-        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
-        json={"amount_cents": 50, "source": "coin_cash"},
+        json={"amount_cents": 100, "source": "coin_cash"},
         headers=_auth(token),
     )
 

tests/test_withdraw_ledger_check.py

@@ -139,8 +139,7 @@ def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
     before = _ledger()
     r = client.post(
         "/api/v1/wallet/withdraw",
-        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
-        json={"amount_cents": 50, "source": "coin_cash"},
+        json={"amount_cents": 200, "source": "coin_cash"},
         headers=_auth(token),
     )
     assert r.status_code == 200, r.text

tests/test_withdraw_tiers.py

@@ -1,186 +0,0 @@
-"""福利页(coin_cash)提现档位规则测试(7-9提现ui对齐)。
-
-规则(2026-07-09 拍板):
-  - 档位 0.1/0.3(新人,历史一次性,免广告)+ 0.5(日3次)/10/20(日1次)。
-  - 计次口径"发起就算":当天创建的单不论最终状态(含被拒)都占名额;新人档任何状态都算用过。
-  - 常规三档每天只能选一个;新人档不参与该互斥,两个新人档同天可各提一次。
-  - invite_cash 无档位概念:tiers 为空、下单不走档位闸(邀请页行为不变)。
-wxpay 调用全部 monkeypatch;现金余额 DB 直灌(同 test_withdraw.py 套路)。
-"""
-from __future__ import annotations
-
-from sqlalchemy import select
-
-from app.db.session import SessionLocal
-from app.models.user import User
-from app.models.wallet import CoinAccount
-from app.repositories import wallet as crud_wallet
-
-
-def _login(client, phone: str) -> str:
-    client.post("/api/v1/auth/sms/send", json={"phone": phone})
-    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
-    assert r.status_code == 200, r.text
-    return r.json()["access_token"]
-
-
-def _auth(token: str) -> dict[str, str]:
-    return {"Authorization": f"Bearer {token}"}
-
-
-def _seed_balances(client, token: str, phone: str, cash: int = 0, invite_cash: int = 0) -> None:
-    client.get("/api/v1/wallet/account", headers=_auth(token))
-    db = SessionLocal()
-    try:
-        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
-        acc = db.get(CoinAccount, user.id)
-        acc.cash_balance_cents = cash
-        acc.invite_cash_balance_cents = invite_cash
-        db.commit()
-    finally:
-        db.close()
-
-
-def _patch_userinfo(monkeypatch, openid: str) -> None:
-    monkeypatch.setattr(
-        "app.integrations.wxpay.code_to_userinfo",
-        lambda code: {"openid": openid, "nickname": "昵称", "avatar_url": None, "raw": {}},
-    )
-
-
-def _reject(bill: str) -> None:
-    """管理员拒绝:退款结清活跃单(便于同用户继续发起下一笔;名额按'发起就算'仍占)。"""
-    db = SessionLocal()
-    try:
-        crud_wallet.reject_withdraw(db, bill, "test")
-    finally:
-        db.close()
-
-
-def _withdraw(client, token: str, cents: int):
-    return client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": cents, "source": "coin_cash"},
-        headers=_auth(token),
-    )
-
-
-def _tiers(client, token: str, source: str = "coin_cash") -> list[dict]:
-    r = client.get(
-        "/api/v1/wallet/withdraw-info", params={"source": source}, headers=_auth(token)
-    )
-    assert r.status_code == 200, r.text
-    return r.json()["tiers"]
-
-
-def test_withdraw_info_tiers_full_and_invite_empty(client, monkeypatch) -> None:
-    """新用户 coin_cash 下发 5 档(新人角标齐);invite_cash tiers 为空。"""
-    token = _login(client, "13800006001")
-    tiers = _tiers(client, token)
-    assert [t["amount_cents"] for t in tiers] == [10, 30, 50, 1000, 2000]
-    assert [t["label"] for t in tiers] == ["0.1", "0.3", "0.5", "10", "20"]
-    assert tiers[0]["badge"] == "新人福利" and tiers[0]["is_newbie"] is True
-    assert tiers[1]["badge"] == "新人福利" and tiers[1]["is_newbie"] is True
-    assert all(t["available"] for t in tiers)
-    assert tiers[2]["remaining_today"] == 3  # 0.5 日 3 次
-    assert _tiers(client, token, source="invite_cash") == []
-
-
-def test_newbie_tiers_independent_and_once_forever(client, monkeypatch) -> None:
-    """0.1 提过(即使被拒)→ 永久消失;同天 0.3 仍可提;新人档不锁常规档。"""
-    _patch_userinfo(monkeypatch, "openid_tier_2")
-    token = _login(client, "13800006002")
-    _seed_balances(client, token, "13800006002", cash=5000)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    r = _withdraw(client, token, 10)  # 0.1 新人档
-    assert r.status_code == 200, r.text
-    _reject(r.json()["out_bill_no"])  # 被拒也算用过("发起就算")
-
-    tiers = _tiers(client, token)
-    amounts = [t["amount_cents"] for t in tiers]
-    assert 10 not in amounts          # 0.1 消失
-    assert 30 in amounts              # 0.3 还在,同天仍可提
-    # 新人档不参与"选一个额度":常规三档全部仍可提
-    regular = {t["amount_cents"]: t for t in tiers if not t["is_newbie"]}
-    assert all(regular[a]["available"] for a in (50, 1000, 2000))
-
-    r = _withdraw(client, token, 30)  # 同天 0.3 照提
-    assert r.status_code == 200, r.text
-    _reject(r.json()["out_bill_no"])
-
-    # 0.1 已用过,重提 → 非法金额(档位已消失)
-    r = _withdraw(client, token, 10)
-    assert r.status_code == 400, r.text
-
-
-def test_regular_daily_select_one_tier(client, monkeypatch) -> None:
-    """当天提过 0.5 → 10/20 置灰 other_tier_selected,下单 409;0.5 还能继续提(3 次内)。"""
-    _patch_userinfo(monkeypatch, "openid_tier_3")
-    token = _login(client, "13800006003")
-    _seed_balances(client, token, "13800006003", cash=10_000)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    r = _withdraw(client, token, 50)
-    assert r.status_code == 200, r.text
-    _reject(r.json()["out_bill_no"])  # 结清活跃单;当天名额仍占("发起就算")
-
-    tiers = {t["amount_cents"]: t for t in _tiers(client, token)}
-    assert tiers[50]["available"] and tiers[50]["remaining_today"] == 2
-    assert not tiers[1000]["available"] and tiers[1000]["disabled_reason"] == "other_tier_selected"
-    assert not tiers[2000]["available"] and tiers[2000]["disabled_reason"] == "other_tier_selected"
-
-    r = _withdraw(client, token, 1000)  # 选一额度互斥 → 409
-    assert r.status_code == 409, r.text
-    assert "今日额度已达上限" in r.json()["detail"]
-
-    r = _withdraw(client, token, 50)  # 0.5 第 2 次照常
-    assert r.status_code == 200, r.text
-
-
-def test_regular_daily_quota_exhausted(client, monkeypatch) -> None:
-    """0.5 日 3 次:第 4 次 409;tiers 显示 quota_exhausted。"""
-    _patch_userinfo(monkeypatch, "openid_tier_4")
-    token = _login(client, "13800006004")
-    _seed_balances(client, token, "13800006004", cash=10_000)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    for _ in range(3):
-        r = _withdraw(client, token, 50)
-        assert r.status_code == 200, r.text
-        _reject(r.json()["out_bill_no"])
-
-    tiers = {t["amount_cents"]: t for t in _tiers(client, token)}
-    assert not tiers[50]["available"]
-    assert tiers[50]["disabled_reason"] == "quota_exhausted"
-    assert tiers[50]["remaining_today"] == 0
-
-    r = _withdraw(client, token, 50)
-    assert r.status_code == 409, r.text
-
-
-def test_non_tier_amount_rejected_for_coin_cash(client, monkeypatch) -> None:
-    """coin_cash 只能提预设档位:任意其他金额 400(防绕过客户端刷)。"""
-    _patch_userinfo(monkeypatch, "openid_tier_5")
-    token = _login(client, "13800006005")
-    _seed_balances(client, token, "13800006005", cash=10_000)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    r = _withdraw(client, token, 123)
-    assert r.status_code == 400, r.text
-
-
-def test_invite_cash_not_gated_by_tiers(client, monkeypatch) -> None:
-    """invite_cash 不走档位闸:非档位金额(200=2元)照常下单——邀请页行为不变。"""
-    _patch_userinfo(monkeypatch, "openid_tier_6")
-    token = _login(client, "13800006006")
-    _seed_balances(client, token, "13800006006", invite_cash=500)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-    assert r.json()["status"] == "reviewing"