优化：修正广告收益看板统计与明细口径

## 修改内容
- 新增短信、一键登录、比价三类风控事件与规则聚合
- 新增管理员忽略、封禁、解封、重置报警和阈值配置接口
- 风险列表返回当前有效限制的 restriction_id，供后台已封禁视图解除
- 在短信/一键登录、比价、任务领奖、提现链路接入风险记录与限制
- 新增通用行为流水、风险事件、主体限制模型及 Alembic 迁移
- 新增本地演示数据脚本与风控测试

## 验证
- ruff check：通过
- Alembic 全新 SQLite upgrade head / downgrade -1：通过
- 风控测试：通过，覆盖封禁列表 restriction_id 与解除链路
- 全量测试：532 passed，8 failed；其中 7 项在干净 origin/main 独立复现，另 1 项独立复跑通过，未发现本分支新增回归

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #174
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>

alembic/versions/drop_withdraw_active_uniq_idx.py

@@ -0,0 +1,37 @@
+"""drop withdraw active-order partial unique index (allow multiple in-flight withdrawals)
+
+Revision ID: drop_withdraw_active_uniq_idx
+Revises: d8dd2106e438
+Create Date: 2026-07-24 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "drop_withdraw_active_uniq_idx"
+down_revision: Union[str, Sequence[str], None] = "d8dd2106e438"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 取消「同一用户同一时刻仅一笔在途提现」:允许 reviewing/pending 并存。
+    # 仅删本索引;姊妹索引 ux_cash_transaction_withdraw_refund_ref(退款幂等)保持不动。
+    op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order")
+
+
+def downgrade() -> None:
+    # 回滚重建分区唯一索引。注意:若届时某用户已有 ≥2 张在途单,重建会因唯一冲突失败——
+    # 属预期的回滚代价(取消限制后本就允许多单),需先人工收敛在途单再回滚。
+    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')"),
+    )

alembic/versions/risk_monitor_generic_tables.py

@@ -0,0 +1,210 @@
+"""add generic behavior, risk incident and subject restriction tables
+
+Revision ID: risk_monitor_generic
+Revises: drop_withdraw_active_uniq_idx
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "risk_monitor_generic"
+down_revision: str | None = "drop_withdraw_active_uniq_idx"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+_RISK_MONITOR_PAGE = "risk-monitor"
+_DEFAULT_RISK_MONITOR_ROLES = ("operator", "tech")
+
+
+def _role_table() -> sa.TableClause:
+    return sa.table(
+        "admin_role",
+        sa.column("name", sa.String),
+        sa.column("pages", _JSON),
+    )
+
+
+def _add_default_role_permissions() -> None:
+    """Grant the new page without replacing any existing role customisation."""
+    role = _role_table()
+    conn = op.get_bind()
+    rows = conn.execute(
+        sa.select(role.c.name, role.c.pages).where(
+            role.c.name.in_(_DEFAULT_RISK_MONITOR_ROLES)
+        )
+    ).all()
+    for name, pages in rows:
+        current_pages = list(pages or [])
+        if _RISK_MONITOR_PAGE not in current_pages:
+            conn.execute(
+                role.update()
+                .where(role.c.name == name)
+                .values(pages=[*current_pages, _RISK_MONITOR_PAGE])
+            )
+
+
+def _remove_default_role_permissions() -> None:
+    role = _role_table()
+    conn = op.get_bind()
+    rows = conn.execute(
+        sa.select(role.c.name, role.c.pages).where(
+            role.c.name.in_(_DEFAULT_RISK_MONITOR_ROLES)
+        )
+    ).all()
+    for name, pages in rows:
+        current_pages = list(pages or [])
+        if _RISK_MONITOR_PAGE in current_pages:
+            conn.execute(
+                role.update()
+                .where(role.c.name == name)
+                .values(
+                    pages=[
+                        page
+                        for page in current_pages
+                        if page != _RISK_MONITOR_PAGE
+                    ]
+                )
+            )
+
+
+def upgrade() -> None:
+    op.create_table(
+        "behavior_event",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("event_type", sa.String(length=64), nullable=False),
+        sa.Column("subject_type", sa.String(length=32), nullable=False),
+        sa.Column("subject_id", sa.String(length=128), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("device_id", sa.String(length=128), nullable=True),
+        sa.Column("device_model", sa.String(length=128), nullable=True),
+        sa.Column("phone", sa.String(length=20), nullable=True),
+        sa.Column("client_ip", sa.String(length=64), nullable=True),
+        sa.Column("outcome", sa.String(length=24), nullable=False),
+        sa.Column("reason", sa.String(length=256), nullable=True),
+        sa.Column("details", sa.JSON(), nullable=True),
+        sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_behavior_event_occurred_at", "behavior_event", ["occurred_at"])
+    op.create_index(
+        "ix_behavior_event_type_time", "behavior_event", ["event_type", "occurred_at"]
+    )
+    op.create_index(
+        "ix_behavior_event_subject_time",
+        "behavior_event",
+        ["subject_type", "subject_id", "occurred_at"],
+    )
+    op.create_index(
+        "ix_behavior_event_user_time", "behavior_event", ["user_id", "occurred_at"]
+    )
+
+    op.create_table(
+        "risk_incident",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("rule_code", sa.String(length=64), nullable=False),
+        sa.Column("event_type", sa.String(length=64), nullable=False),
+        sa.Column("subject_type", sa.String(length=32), nullable=False),
+        sa.Column("subject_id", sa.String(length=128), nullable=False),
+        sa.Column("window_key", sa.String(length=64), nullable=False),
+        sa.Column("window_start", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("window_end", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("first_event_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("triggered_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("last_event_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("event_count", sa.Integer(), nullable=False),
+        sa.Column("status", sa.String(length=24), nullable=False),
+        sa.Column("action_reason", sa.String(length=256), nullable=True),
+        sa.Column("handled_by", sa.Integer(), nullable=True),
+        sa.Column("handled_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("details", sa.JSON(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "rule_code",
+            "subject_type",
+            "subject_id",
+            "window_key",
+            name="uq_risk_incident_rule_subject_window",
+        ),
+    )
+    op.create_index(
+        "ix_risk_incident_rule_status",
+        "risk_incident",
+        ["rule_code", "status", "triggered_at"],
+    )
+    op.create_index(
+        "ix_risk_incident_subject",
+        "risk_incident",
+        ["subject_type", "subject_id", "triggered_at"],
+    )
+
+    op.create_table(
+        "subject_restriction",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("subject_type", sa.String(length=32), nullable=False),
+        sa.Column("subject_id", sa.String(length=128), nullable=False),
+        sa.Column("scope", sa.String(length=32), nullable=False),
+        sa.Column("active", sa.Boolean(), server_default=sa.true(), nullable=False),
+        sa.Column("reason", sa.String(length=256), nullable=True),
+        sa.Column("incident_id", sa.Integer(), nullable=True),
+        sa.Column("created_by", sa.Integer(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.Column("revoked_by", sa.Integer(), nullable=True),
+        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("details", sa.JSON(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "subject_type",
+            "subject_id",
+            "scope",
+            name="uq_subject_restriction_subject_scope",
+        ),
+    )
+    op.create_index(
+        "ix_subject_restriction_lookup",
+        "subject_restriction",
+        ["subject_type", "subject_id", "scope", "active"],
+    )
+    _add_default_role_permissions()
+
+
+def downgrade() -> None:
+    _remove_default_role_permissions()
+    op.drop_index("ix_subject_restriction_lookup", table_name="subject_restriction")
+    op.drop_table("subject_restriction")
+    op.drop_index("ix_risk_incident_subject", table_name="risk_incident")
+    op.drop_index("ix_risk_incident_rule_status", table_name="risk_incident")
+    op.drop_table("risk_incident")
+    op.drop_index("ix_behavior_event_user_time", table_name="behavior_event")
+    op.drop_index("ix_behavior_event_subject_time", table_name="behavior_event")
+    op.drop_index("ix_behavior_event_type_time", table_name="behavior_event")
+    op.drop_index("ix_behavior_event_occurred_at", table_name="behavior_event")
+    op.drop_table("behavior_event")

app/admin/main.py

@@ -17,6 +17,7 @@ from app.admin.routers.ad_audit import router as ad_audit_router
 from app.admin.routers.ad_config import router as ad_config_router
 from app.admin.routers.ad_revenue import router as ad_revenue_router
 from app.admin.routers.admins import router as admins_router
+from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.audit import router as audit_router
 from app.admin.routers.auth import router as auth_router
 from app.admin.routers.comparison import router as comparison_router
@@ -25,8 +26,6 @@ from app.admin.routers.coupon_data import router as coupon_data_router
 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
@@ -34,7 +33,9 @@ from app.admin.routers.guide_video import router as guide_video_router
 from app.admin.routers.huawei_review import router as huawei_review_router
 from app.admin.routers.onboarding import router as onboarding_router
 from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
+from app.admin.routers.ops_stat_config import router as ops_stat_config_router
 from app.admin.routers.price_report import router as price_report_router
+from app.admin.routers.risk_monitor import router as risk_monitor_router
 from app.admin.routers.roles import router as roles_router
 from app.admin.routers.users import router as users_router
 from app.admin.routers.wallet import router as wallet_router
@@ -101,6 +102,7 @@ admin_app.include_router(onboarding_router)
 admin_app.include_router(wallet_router)
 admin_app.include_router(withdraw_router)
 admin_app.include_router(price_report_router)
+admin_app.include_router(risk_monitor_router)
 admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
 admin_app.include_router(analytics_health_router)

app/admin/permissions.py

@@ -34,6 +34,7 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "users", "label": "用户管理"},
     ]},
     {"group": "监控审计", "pages": [
+        {"key": "risk-monitor", "label": "风控监控"},
         {"key": "device-liveness", "label": "设备存活"},
         {"key": "analytics-health", "label": "埋点成功率"},
         {"key": "event-logs", "label": "埋点日志"},
@@ -55,13 +56,13 @@ BUILTIN_ROLES: list[dict] = [
     {"name": SUPER_ADMIN_ROLE, "label": "管理员", "pages": []},
     {"name": "operator", "label": "运营", "pages": [
         "dashboard", "coupon-data", "ad-revenue-report", "comparison-records",
-        "cps", "device-liveness", "price-reports", "feedbacks", "huawei-review",
+        "cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
     ]},
     {"name": "finance", "label": "财务", "pages": [
         "dashboard", "ad-revenue-report", "cps", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
-        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
+        "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
         "event-logs", "audit-logs",
     ]},
 ]

app/admin/repositories/ad_audit.py

@@ -12,10 +12,11 @@
 """
 from __future__ import annotations
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
+from app.models.ad_ecpm import AdEcpmRecord
 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
@@ -55,10 +56,22 @@ def _reward_video_rows(
     if user_id is not None:
         stmt = stmt.where(AdRewardRecord.user_id == user_id)
 
+    records = list(db.execute(stmt).scalars())
+    # S2S 发奖回调本身不携带实际填充的 ADN/底层 rit；按客户端在展示时上报的
+    # ad_session_id 回填。这样“纯发奖”行也能在运营后台追溯到真实广告网络。
+    session_ids = {rec.ad_session_id for rec in records if rec.ad_session_id}
+    impression_by_session = {
+        (rec.user_id, rec.ad_session_id): rec
+        for rec in db.execute(
+            select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
+        ).scalars()
+    } if session_ids else {}
+
     # 用本日之前的累计份数做起点,当日 granted 在其上继续递增 → 与 _granted_cumulative+1 对齐
     granted_n: dict[int, int] = _prior_granted_counts(db, date=date, user_id=user_id)
     rows: list[dict] = []
-    for rec in db.execute(stmt).scalars():
+    for rec in records:
+        impression = impression_by_session.get((rec.user_id, rec.ad_session_id))
         if rec.status == "granted":
             nth = granted_n.get(rec.user_id, 0) + 1
             granted_n[rec.user_id] = nth
@@ -68,6 +81,8 @@ def _reward_video_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": impression.adn if impression is not None else None,
+                "slot_id": impression.slot_id if impression is not None else None,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -90,6 +105,8 @@ def _reward_video_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": impression.adn if impression is not None else None,
+                "slot_id": impression.slot_id if impression is not None else None,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -149,6 +166,81 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
     return True
 
 
+def _nonblank(value: str | None) -> str | None:
+    value = value.strip() if value else None
+    return value or None
+
+
+def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | None]:
+    """仅在候选展示记录指向唯一 ADN 时回填来源，绝不把一次多广告流程猜成某一个网络。"""
+    adns = {_nonblank(record.adn) for record in records}
+    adns.discard(None)
+    if len(adns) != 1:
+        return None, None
+
+    slots = {_nonblank(record.slot_id) for record in records}
+    slots.discard(None)
+    return next(iter(adns)), next(iter(slots)) if len(slots) == 1 else None
+
+
+def _feed_source_fallbacks(
+    db: Session, records: list[AdFeedRewardRecord]
+) -> tuple[dict[tuple[int, str], tuple[str | None, str | None]], dict[tuple[int, str, str], tuple[str | None, str | None]]]:
+    """构建信息流来源回填索引。
+
+    新客户端会把 ADN 直接随 feed-reward 上报；旧记录可能缺失。展示收益记录的
+    ``ad_session_id`` 是每条 impressionId，而发奖记录保留的是整场会话 ID，因此先按
+    会话精确匹配；匹配不到时仅允许按 ``user + trace_id + 原始 eCPM`` 回填，且候选 ADN
+    必须唯一。trace 内存在多个网络时保持空值，避免错误归因。
+    """
+    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
+    trace_ids = {record.trace_id for record in records if record.trace_id}
+    if not session_ids and not trace_ids:
+        return {}, {}
+
+    filters = []
+    if session_ids:
+        filters.append(AdEcpmRecord.ad_session_id.in_(session_ids))
+    if trace_ids:
+        filters.append(AdEcpmRecord.trace_id.in_(trace_ids))
+    impressions = list(db.execute(select(AdEcpmRecord).where(or_(*filters))).scalars())
+
+    by_session: dict[tuple[int, str], list[AdEcpmRecord]] = {}
+    by_trace_ecpm: dict[tuple[int, str, str], list[AdEcpmRecord]] = {}
+    for impression in impressions:
+        if impression.ad_session_id:
+            by_session.setdefault((impression.user_id, impression.ad_session_id), []).append(impression)
+        if impression.trace_id:
+            by_trace_ecpm.setdefault(
+                (impression.user_id, impression.trace_id, impression.ecpm_raw), []
+            ).append(impression)
+
+    return (
+        {key: _unique_ad_source(value) for key, value in by_session.items()},
+        {key: _unique_ad_source(value) for key, value in by_trace_ecpm.items()},
+    )
+
+
+def _feed_source(
+    record: AdFeedRewardRecord,
+    *,
+    by_session: dict[tuple[int, str], tuple[str | None, str | None]],
+    by_trace_ecpm: dict[tuple[int, str, str], tuple[str | None, str | None]],
+) -> tuple[str | None, str | None]:
+    """取得本条发奖广告的真实来源；无唯一证据时返回原始空值。"""
+    adn, slot_id = _nonblank(record.adn), _nonblank(record.slot_id)
+    if adn and slot_id:
+        return adn, slot_id
+
+    candidate = by_session.get((record.user_id, record.ad_session_id or ""))
+    if candidate is None and record.trace_id:
+        candidate = by_trace_ecpm.get((record.user_id, record.trace_id, record.ecpm_raw))
+    if candidate is None:
+        return adn, slot_id
+    candidate_adn, candidate_slot_id = candidate
+    return adn or candidate_adn, slot_id or candidate_slot_id
+
+
 def _feed_rows(
     db: Session, *, date: str, user_id: int | None, scene: str | None = None
 ) -> list[dict]:
@@ -167,11 +259,17 @@ def _feed_rows(
     if user_id is not None:
         stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
 
+    records = list(db.execute(stmt).scalars())
+    by_session, by_trace_ecpm = _feed_source_fallbacks(db, records)
+
     # 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
     granted_count: dict[int, int] = _feed_prior_granted_count(db, date=date, user_id=user_id)
     rows: list[dict] = []
-    for rec in db.execute(stmt).scalars():
+    for rec in records:
         keep = _feed_scene_matches(rec, scene)  # 累计照常推进,这里只决定是否展示本行
+        adn, slot_id = _feed_source(
+            rec, by_session=by_session, by_trace_ecpm=by_trace_ecpm
+        )
         if rec.status == "granted":
             # 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
             # nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
@@ -188,6 +286,8 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": adn,
+                "slot_id": slot_id,
                 "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
@@ -214,6 +314,8 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": adn,
+                "slot_id": slot_id,
                 "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,

app/admin/repositories/ad_revenue.py

@@ -22,7 +22,7 @@ report_date / reward_date 归日。
 """
 from __future__ import annotations
 
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime, time, timedelta
 from datetime import date as _date
 
 from sqlalchemy import select
@@ -81,9 +81,9 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 # ad_feed_reward_record,由 audit 内部按 ad_type 区分(feed 含历史 NULL,draw 仅 ad_type=="draw")。
 _AUDIT_SCENES = {"reward_video", "feed", "draw"}
 
-# 激励视频未满足有效播放条件时不计客户端预估收益。客户端仍会在 onAdShow
-# 上报 eCPM，随后才在关闭时补报以下终态，因此必须在展示/发奖合并后修正收益。
-_ZERO_REVENUE_REWARD_VIDEO_STATUSES = frozenset({"closed_early", "too_short"})
+# GroMore 官方说明第三方 ADN 的 Reporting API 最晚约 13:50 更新。只有 D+1 14:00
+# 之后完成的同步才标记为「API 同步窗口完成」；这不代表覆盖全部 ADN 或最终结算。
+_PANGLE_API_FINAL_SYNC_TIME = time(hour=14)
 
 
 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
@@ -96,7 +96,32 @@ _REWARD_DETAIL_KEYS = (
 
 def _reward_detail(row: dict) -> dict:
     """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
-    return {k: row[k] for k in _REWARD_DETAIL_KEYS}
+    detail = {k: row[k] for k in _REWARD_DETAIL_KEYS}
+    # 发奖明细必须保留自己的广告网络，不能复用整场聚合父行的来源：
+    # 同一次比价/领券可能先后由不同 ADN 填充。
+    detail["adn"] = row.get("adn")
+    detail["slot_id"] = row.get("slot_id")
+    return detail
+
+
+def _as_cn(dt: datetime) -> datetime:
+    """数据库 synced_at → 北京时间；SQLite naive 值按 UTC 处理。"""
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return dt.astimezone(rewards.CN_TZ)
+
+
+def _pangle_api_day_complete(day: str, aggregate: dict) -> bool:
+    """某天 API 收益是否已在 D+1 14:00 后同步（仅表示同步窗口完成）。"""
+    synced_at = aggregate.get("synced_at")
+    if aggregate.get("api_revenue_yuan") is None or synced_at is None:
+        return False
+    cutoff = datetime.combine(
+        _date.fromisoformat(day) + timedelta(days=1),
+        _PANGLE_API_FINAL_SYNC_TIME,
+        tzinfo=rewards.CN_TZ,
+    )
+    return _as_cn(synced_at) >= cutoff
 
 
 def ad_revenue_report(
@@ -186,12 +211,10 @@ def ad_revenue_report(
             "has_impression": True,
             "impressions": 1,
             "ecpm": rec.ecpm_raw,
-            # 单次展示收益(元)= eCPM元 ÷ 1000(每千次→单次)。eCPM 先钳到 AD_ECPM_MAX_FEN(¥500 CPM)
-            # 再折收益,与发奖口径 [rewards.calculate_ad_reward_coin] 一致(2026-06-29 修:原裸 parse_ecpm_yuan
-            # 不钳,伪造/异常天价 eCPM 会把报表预估收益冲到任意大;金币侧已钳、收益侧漏钳)。
-            "revenue_yuan": round(
-                min(rewards.parse_ecpm_yuan(rec.ecpm_raw), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0, 6,
-            ),
+            # 客户端 SDK 展示预估收益(元)= 后端留存 getEcpm 元/千次 ÷ 1000。
+            # 这里不能复用发奖防作弊的 ¥500 CPM 钳顶：钳顶只限制金币成本，不改变广告已产生的
+            # 收入估值。onAdShow 已发生即计展示收入，是否看满只影响发奖，不影响广告收入。
+            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0, 6),
             "adn": rec.adn,
             "slot_id": rec.slot_id,
             "sub_rewards": [],
@@ -206,11 +229,6 @@ def ad_revenue_report(
                 "matched": bool(rwd["matched"]),
                 "reward_detail": _reward_detail(rwd),
             })
-            if (
-                rec.ad_type == "reward_video"
-                and rwd["status"] in _ZERO_REVENUE_REWARD_VIDEO_STATUSES
-            ):
-                ev["revenue_yuan"] = 0.0
         else:
             # 纯展示(信息流逐条展示、激励视频缺发奖记录):不计对账,matched=True。
             ev.update({
@@ -244,8 +262,8 @@ def ad_revenue_report(
                 "impressions": 0,
                 "ecpm": row["ecpm"],
                 "revenue_yuan": 0.0,
-                "adn": None,
-                "slot_id": None,
+                "adn": row.get("adn"),
+                "slot_id": row.get("slot_id"),
                 "has_reward": True,
                 "status": row["status"],
                 "expected_coin": int(row["expected_coin"]),
@@ -271,10 +289,10 @@ def ad_revenue_report(
         # 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条
         ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")]
         avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm")
-        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进
+        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算)。只放进
         # row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。
         row_revenue = round(sum(
-            min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
+            rewards.parse_ecpm_yuan(g["ecpm"]) / 1000.0
             for g in group if g.get("ecpm")
         ), 6)
         events.append({
@@ -364,13 +382,15 @@ def ad_revenue_report(
         for d in sorted(daily_map.values(), key=lambda x: x["date"])
     ]
 
-    # 穿山甲后台收益(GroMore 数据 API,T+1 入库 ad_pangle_daily_revenue):汇总 + 按天趋势级展示,
+    # GroMore 排序价预估 / ADN Reporting API 收益(T+1 入库):汇总 + 按天趋势级展示,
     # 与上面客户端自报 eCPM 折算的预估并列对照(看 gap)。穿山甲数据**无用户/场景/类型维度**,故仅在
     # 「全量视图」(未按 user_id / ad_type / feed_scene 过滤)给值;一旦带这些过滤,穿山甲数无法对应口径
     # → 置 None,前端显示「-」并提示。逐条事件行不动(仍是客户端预估)。
     pangle_filterable = user_id is None and ad_type is None and feed_scene is None
     total_pangle_revenue_yuan: float | None = None
     total_pangle_api_revenue_yuan: float | None = None
+    pangle_api_revenue_complete = False
+    pangle_latest_synced_at: datetime | None = None
     if pangle_filterable:
         pangle_aggs = ad_pangle_revenue.aggregate_by_date(
             db,
@@ -388,6 +408,12 @@ def ad_revenue_report(
             total_pangle_revenue_yuan = round(sum(a["revenue_yuan"] for a in pangle_aggs), 6)
             api_vals = [a["api_revenue_yuan"] for a in pangle_aggs if a["api_revenue_yuan"] is not None]
             total_pangle_api_revenue_yuan = round(sum(api_vals), 6) if api_vals else None
+            sync_times = [a["synced_at"] for a in pangle_aggs if a["synced_at"] is not None]
+            pangle_latest_synced_at = max(sync_times) if sync_times else None
+            pangle_api_revenue_complete = all(
+                day in by_date and _pangle_api_day_complete(day, by_date[day])
+                for day in _date_range(date_from, date_to)
+            )
 
     # 按小时汇总(全量,不受分页 limit/offset 影响):供前端按小时趋势图(单日 granularity=hour 时用)。
     # 只在 by_hour 下聚合(此时每个 event 带 hour);否则空。前端按天趋势仍用 daily。
@@ -412,41 +438,50 @@ def ad_revenue_report(
             for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
         ]
 
-    # 分广告类型小计(按 ad_type:展示条数 + 预估收益;eCPM 由前端用 收益÷展示×1000 算)。
-    # 基于全量(已按 feed_scene 过滤)events;前端只取 draw / reward_video 两类展示。
-    type_map: dict[str, dict] = {}
-    for e in events:
-        t = type_map.get(e["ad_type"])
-        if t is None:
-            t = {"impressions": 0, "revenue_yuan": 0.0}
-            type_map[e["ad_type"]] = t
-        t["impressions"] += e["impressions"]
-        t["revenue_yuan"] += e["revenue_yuan"]
-    type_stats = {
-        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
-        for k, v in type_map.items()
-    }
+    def _aggregate_stats(bucket_of) -> dict[str, dict]:
+        """按展示事件聚合收益 / 加权 SDK eCPM，避免前端漏合并历史类型。"""
+        stat_map: dict[str, dict] = {}
+        for e in events:
+            bucket = bucket_of(e)
+            if bucket is None:
+                continue
+            stat = stat_map.setdefault(bucket, {
+                "impressions": 0,
+                "revenue_yuan": 0.0,
+                "ecpm_fen_sum": 0.0,
+            })
+            impressions = int(e["impressions"])
+            stat["impressions"] += impressions
+            stat["revenue_yuan"] += e["revenue_yuan"]
+            # eCPM 必须以每次真实展示为权重；纯发奖父行 impressions=0，不能参与分母或均值。
+            stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(e["ecpm"]) * impressions
+        return {
+            key: {
+                "impressions": value["impressions"],
+                "revenue_yuan": round(value["revenue_yuan"], 6),
+                "ecpm_yuan": round(
+                    value["ecpm_fen_sum"] / value["impressions"] / 100.0,
+                    6,
+                ) if value["impressions"] else 0.0,
+            }
+            for key, value in stat_map.items()
+        }
 
-    # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
-    # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
-    # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
-    # 改为服务端在全量上聚合下发(也顺带不受 limit 分页截断影响)。feed_scene 为空(激励视频 /
-    # 旧数据)不计入任何场景桶。
-    scene_map: dict[str, dict] = {}
-    for e in events:
-        sc = e.get("feed_scene")
-        if not sc:
-            continue
-        s = scene_map.get(sc)
-        if s is None:
-            s = {"impressions": 0, "revenue_yuan": 0.0}
-            scene_map[sc] = s
-        s["impressions"] += e["impressions"]
-        s["revenue_yuan"] += e["revenue_yuan"]
-    scene_stats = {
-        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
-        for k, v in scene_map.items()
-    }
+    # 原始 ad_type 小计，供明细筛选和排查使用。
+    type_stats = _aggregate_stats(lambda e: e["ad_type"])
+    # 经营看板使用的规范分类：Draw 包含历史 feed；看视频包含福利与提现视频。
+    # 这两个集合与筛选逻辑保持一致，避免只取 draw / reward_video 而漏算历史或提现数据。
+    category_stats = _aggregate_stats(
+        lambda e: (
+            "draw" if e["ad_type"] in {"draw", "feed"}
+            else "video" if e["ad_type"] in {"reward_video", "withdrawal_video"}
+            else None
+        )
+    )
+
+    # 分场景小计，同 type_stats 基于全量 events，供数据大盘「领券广告 / 比价广告」卡使用。
+    # feed_scene 为空的激励视频 / 历史数据不计入任何场景桶。
+    scene_stats = _aggregate_stats(lambda e: e.get("feed_scene"))
 
     # DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
     # 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
@@ -471,9 +506,11 @@ def ad_revenue_report(
         "truncated": len(main_rows) > offset + limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
-        # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。
+        # GroMore 排序价预估 + ADN Reporting API 收益;非全量视图或无数据为 None。
         "total_pangle_revenue_yuan": total_pangle_revenue_yuan,
         "total_pangle_api_revenue_yuan": total_pangle_api_revenue_yuan,
+        "pangle_api_revenue_complete": pangle_api_revenue_complete,
+        "pangle_latest_synced_at": pangle_latest_synced_at,
         "pangle_revenue_available": total_pangle_revenue_yuan is not None,
         "total_expected_coin": total_expected_coin,
         "total_actual_coin": total_actual_coin,
@@ -481,6 +518,7 @@ def ad_revenue_report(
         "daily": daily,
         "hourly": hourly,
         "type_stats": type_stats,
+        "category_stats": category_stats,
         "scene_stats": scene_stats,
         "dau": dau,
         "items": main_rows[offset:offset + limit],

app/admin/repositories/risk_monitor.py

@@ -0,0 +1,545 @@
+"""风控监控读模型：聚合通用行为流水、风险事件与比价记录。"""
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import desc, func, select
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.models.comparison import ComparisonRecord
+from app.models.device import DeviceLiveness
+from app.models.risk import BehaviorEvent, RiskIncident
+from app.models.user import User
+from app.repositories import risk as risk_repo
+
+KIND_TO_RULE = {
+    "sms": risk_repo.RULE_SMS_HOURLY,
+    "oneclick": risk_repo.RULE_ONECLICK_DAILY,
+    "compare": risk_repo.RULE_COMPARE_DAILY,
+}
+RULE_TO_KIND = {rule: kind for kind, rule in KIND_TO_RULE.items()}
+
+
+def _day_bounds(now: datetime | None = None) -> tuple[str, datetime, datetime]:
+    current = (now or datetime.now(UTC)).astimezone(risk_repo.CN_TZ)
+    start = current.replace(hour=0, minute=0, second=0, microsecond=0)
+    end = start + timedelta(days=1)
+    return (
+        start.strftime("%Y-%m-%d"),
+        start.astimezone(UTC),
+        end.astimezone(UTC),
+    )
+
+
+def _compare_day_bounds(now: datetime | None = None) -> tuple[datetime, datetime]:
+    current = (now or datetime.now(UTC)).astimezone(risk_repo.CN_TZ).replace(
+        tzinfo=None
+    )
+    start = current.replace(hour=0, minute=0, second=0, microsecond=0)
+    return start, start + timedelta(days=1)
+
+
+def sync_today_compare_incidents(db: Session, now: datetime | None = None) -> None:
+    risk_repo.reconcile_compare_rule(db, at=now)
+
+
+def reset_open_alerts(
+    db: Session, *, admin_id: int, reset_at: datetime
+) -> dict[str, int]:
+    """将三类待处理报警归零，并从 reset_at 开始重新累计。"""
+    rule_codes = tuple(KIND_TO_RULE.values())
+    incidents = list(
+        db.scalars(
+            select(RiskIncident).where(
+                RiskIncident.rule_code.in_(rule_codes),
+                RiskIncident.status == "open",
+            )
+        ).all()
+    )
+    counts = {kind: 0 for kind in KIND_TO_RULE}
+    for incident in incidents:
+        incident.status = "resolved"
+        incident.action_reason = risk_repo.MANUAL_RESET_REASON
+        incident.handled_by = admin_id
+        incident.handled_at = reset_at
+        counts[RULE_TO_KIND[incident.rule_code]] += 1
+    risk_repo.reset_rule_baselines(
+        db,
+        rule_codes=rule_codes,
+        reset_at=reset_at,
+        admin_id=admin_id,
+    )
+    return counts
+
+
+def summary(db: Session, now: datetime | None = None) -> dict:
+    date_key, start, end = _day_bounds(now)
+    compare_start, compare_end = _compare_day_bounds(now)
+    current = now or datetime.now(UTC)
+    risk_repo.reconcile_behavior_rule(
+        db, rule_code=risk_repo.RULE_SMS_HOURLY, at=current, commit=False
+    )
+    risk_repo.reconcile_behavior_rule(
+        db, rule_code=risk_repo.RULE_ONECLICK_DAILY, at=current, commit=False
+    )
+    risk_repo.reconcile_compare_rule(db, at=current, commit=False)
+    db.commit()
+    sms_threshold = risk_repo.get_rule_threshold(db, risk_repo.RULE_SMS_HOURLY)
+    oneclick_threshold = risk_repo.get_rule_threshold(
+        db, risk_repo.RULE_ONECLICK_DAILY
+    )
+    compare_threshold = risk_repo.get_rule_threshold(db, risk_repo.RULE_COMPARE_DAILY)
+
+    def _behavior_total(event_type: str, outcomes: tuple[str, ...]) -> int:
+        return int(
+            db.scalar(
+                select(func.count(BehaviorEvent.id)).where(
+                    BehaviorEvent.event_type == event_type,
+                    BehaviorEvent.outcome.in_(outcomes),
+                    BehaviorEvent.occurred_at >= start,
+                    BehaviorEvent.occurred_at < end,
+                )
+            )
+            or 0
+        )
+
+    compare_total = int(
+        db.scalar(
+            select(func.count(ComparisonRecord.id)).where(
+                ComparisonRecord.created_at >= compare_start,
+                ComparisonRecord.created_at < compare_end,
+            )
+        )
+        or 0
+    )
+
+    def _open_count(rule_code: str) -> int:
+        return int(
+            db.scalar(
+                select(func.count(RiskIncident.id)).where(
+                    RiskIncident.rule_code == rule_code,
+                    RiskIncident.status == "open",
+                    RiskIncident.window_key.like(f"{date_key}%"),
+                )
+            )
+            or 0
+        )
+
+    return {
+        "date": date_key,
+        "updated_at": now or datetime.now(UTC),
+        "cards": [
+            {
+                "kind": "sms",
+                "alert_subject_count": _open_count(risk_repo.RULE_SMS_HOURLY),
+                "today_total": _behavior_total(risk_repo.EVENT_SMS_SEND, ("success",)),
+                "threshold": sms_threshold,
+                "rule_text": f"单设备 1 小时 ≥ {sms_threshold} 次",
+            },
+            {
+                "kind": "oneclick",
+                "alert_subject_count": _open_count(risk_repo.RULE_ONECLICK_DAILY),
+                "today_total": _behavior_total(
+                    risk_repo.EVENT_ONECLICK_LOGIN, ("success", "failed")
+                ),
+                "threshold": oneclick_threshold,
+                "rule_text": f"单设备当日 ≥ {oneclick_threshold} 次",
+            },
+            {
+                "kind": "compare",
+                "alert_subject_count": _open_count(risk_repo.RULE_COMPARE_DAILY),
+                "today_total": compare_total,
+                "threshold": compare_threshold,
+                "rule_text": f"单账户当日 ≥ {compare_threshold} 次",
+            },
+        ],
+    }
+
+
+def _latest_device_model(db: Session, subject_id: str) -> str | None:
+    return db.execute(
+        select(BehaviorEvent.device_model)
+        .where(
+            BehaviorEvent.subject_type == "device",
+            BehaviorEvent.subject_id == subject_id,
+            BehaviorEvent.device_model.is_not(None),
+            BehaviorEvent.device_model != "",
+        )
+        .order_by(BehaviorEvent.occurred_at.desc(), BehaviorEvent.id.desc())
+        .limit(1)
+    ).scalar_one_or_none()
+
+
+def _first_used_at(db: Session, subject_id: str) -> datetime | None:
+    behavior_first = db.scalar(
+        select(func.min(BehaviorEvent.occurred_at)).where(
+            BehaviorEvent.subject_type == "device",
+            BehaviorEvent.subject_id == subject_id,
+        )
+    )
+    registered_first = db.scalar(
+        select(func.min(DeviceLiveness.created_at)).where(
+            DeviceLiveness.device_id == subject_id
+        )
+    )
+    observed = [value for value in (behavior_first, registered_first) if value is not None]
+    if not observed:
+        return None
+
+    def _utc(value: datetime) -> datetime:
+        return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+
+    return min(observed, key=_utc)
+
+
+def _common_device(db: Session, user_id: int, start: datetime, end: datetime) -> str | None:
+    return db.execute(
+        select(ComparisonRecord.device_id)
+        .where(
+            ComparisonRecord.user_id == user_id,
+            ComparisonRecord.device_id.is_not(None),
+            ComparisonRecord.created_at >= start,
+            ComparisonRecord.created_at < end,
+        )
+        .group_by(ComparisonRecord.device_id)
+        .order_by(func.count(ComparisonRecord.id).desc(), ComparisonRecord.device_id.asc())
+        .limit(1)
+    ).scalar_one_or_none()
+
+
+def list_incidents(
+    db: Session,
+    *,
+    kind: str,
+    status: str = "open",
+    limit: int = 20,
+    cursor: int = 0,
+) -> tuple[list[dict], int | None, int]:
+    rule_code = KIND_TO_RULE[kind]
+    stmt = select(RiskIncident).where(RiskIncident.rule_code == rule_code)
+    count_stmt = select(func.count(RiskIncident.id)).where(
+        RiskIncident.rule_code == rule_code
+    )
+    if status != "all":
+        stmt = stmt.where(RiskIncident.status == status)
+        count_stmt = count_stmt.where(RiskIncident.status == status)
+    total = int(db.scalar(count_stmt) or 0)
+    incidents = list(
+        db.scalars(
+            stmt.order_by(desc(RiskIncident.triggered_at), desc(RiskIncident.id))
+            .offset(cursor)
+            .limit(limit)
+        ).all()
+    )
+
+    user_ids = [int(i.subject_id) for i in incidents if i.subject_type == "user"]
+    users = {
+        u.id: u
+        for u in db.scalars(select(User).where(User.id.in_(user_ids))).all()
+    } if user_ids else {}
+
+    items: list[dict] = []
+    for incident in incidents:
+        item = {
+            "incident_id": incident.id,
+            "kind": kind,
+            "subject_type": incident.subject_type,
+            "subject_id": incident.subject_id,
+            "window_start": incident.window_start,
+            "window_end": incident.window_end,
+            "first_event_at": incident.first_event_at,
+            "triggered_at": incident.triggered_at,
+            "last_event_at": incident.last_event_at,
+            "event_count": incident.event_count,
+            "status": incident.status,
+        }
+        if incident.subject_type == "device":
+            restriction = risk_repo.get_active_restriction(
+                db,
+                subject_type="device",
+                subject_id=incident.subject_id,
+                scope=risk_repo.SCOPE_AUTH_DEVICE,
+            )
+            item["device_model"] = _latest_device_model(db, incident.subject_id)
+            item["first_used_at"] = _first_used_at(db, incident.subject_id)
+            item["restricted"] = restriction is not None
+            item["restriction_id"] = restriction.id if restriction else None
+        else:
+            uid = int(incident.subject_id)
+            user = users.get(uid)
+            restriction = risk_repo.get_active_restriction(
+                db,
+                subject_type="user",
+                subject_id=incident.subject_id,
+                scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+            )
+            item.update(
+                {
+                    "user_id": uid,
+                    "phone": user.phone if user else None,
+                    "registered_at": user.created_at if user else None,
+                    "common_device_id": _common_device(
+                        db, uid, incident.window_start, incident.window_end
+                    ),
+                    "restricted": restriction is not None,
+                    "restriction_id": restriction.id if restriction else None,
+                }
+            )
+        items.append(item)
+    next_cursor = cursor + len(items) if cursor + len(items) < total else None
+    return items, next_cursor, total
+
+
+def _page_events(
+    db: Session,
+    incident: RiskIncident,
+    *,
+    event_type: str,
+    outcomes: tuple[str, ...] | None,
+    limit: int,
+    cursor: int,
+) -> tuple[list[BehaviorEvent], int]:
+    filters = [
+        BehaviorEvent.event_type == event_type,
+        BehaviorEvent.subject_type == incident.subject_type,
+        BehaviorEvent.subject_id == incident.subject_id,
+        BehaviorEvent.occurred_at >= incident.window_start,
+        BehaviorEvent.occurred_at < incident.window_end,
+    ]
+    if outcomes:
+        filters.append(BehaviorEvent.outcome.in_(outcomes))
+    total = int(db.scalar(select(func.count(BehaviorEvent.id)).where(*filters)) or 0)
+    rows = list(
+        db.scalars(
+            select(BehaviorEvent)
+            .where(*filters)
+            .order_by(BehaviorEvent.occurred_at.desc(), BehaviorEvent.id.desc())
+            .offset(cursor)
+            .limit(limit)
+        ).all()
+    )
+    return rows, total
+
+
+def _user_map(db: Session, user_ids: set[int]) -> dict[int, User]:
+    if not user_ids:
+        return {}
+    return {u.id: u for u in db.scalars(select(User).where(User.id.in_(user_ids))).all()}
+
+
+def _distinct_event_users(
+    db: Session,
+    incident: RiskIncident,
+    *,
+    event_type: str,
+    outcomes: tuple[str, ...],
+) -> int:
+    return int(
+        db.scalar(
+            select(func.count(func.distinct(BehaviorEvent.user_id))).where(
+                BehaviorEvent.event_type == event_type,
+                BehaviorEvent.subject_type == incident.subject_type,
+                BehaviorEvent.subject_id == incident.subject_id,
+                BehaviorEvent.outcome.in_(outcomes),
+                BehaviorEvent.user_id.is_not(None),
+                BehaviorEvent.occurred_at >= incident.window_start,
+                BehaviorEvent.occurred_at < incident.window_end,
+            )
+        )
+        or 0
+    )
+
+
+def _sms_distinct_accounts(db: Session, incident: RiskIncident) -> int:
+    """统计整个告警窗口内，确实完成本设备短信验证的不同账号数。"""
+    send_rows = db.execute(
+        select(BehaviorEvent.phone, BehaviorEvent.occurred_at).where(
+            BehaviorEvent.event_type == risk_repo.EVENT_SMS_SEND,
+            BehaviorEvent.subject_type == incident.subject_type,
+            BehaviorEvent.subject_id == incident.subject_id,
+            BehaviorEvent.outcome == "success",
+            BehaviorEvent.phone.is_not(None),
+            BehaviorEvent.occurred_at >= incident.window_start,
+            BehaviorEvent.occurred_at < incident.window_end,
+        )
+    ).all()
+    phones = {phone for phone, _ in send_rows}
+    if not phones:
+        return 0
+    login_rows = db.execute(
+        select(
+            BehaviorEvent.user_id,
+            BehaviorEvent.phone,
+            BehaviorEvent.occurred_at,
+        )
+        .where(
+            BehaviorEvent.event_type == risk_repo.EVENT_SMS_LOGIN,
+            BehaviorEvent.subject_id == incident.subject_id,
+            BehaviorEvent.phone.in_(phones),
+            BehaviorEvent.user_id.is_not(None),
+            BehaviorEvent.outcome == "success",
+            BehaviorEvent.occurred_at >= incident.window_start,
+            BehaviorEvent.occurred_at
+            < incident.window_end + timedelta(seconds=settings.SMS_CODE_TTL_SEC),
+        )
+        .order_by(BehaviorEvent.occurred_at.asc(), BehaviorEvent.id.asc())
+    ).all()
+    matched_users = {
+        user_id
+        for user_id, phone, login_at in login_rows
+        if any(
+            sent_phone == phone
+            and sent_at <= login_at
+            and login_at
+            <= sent_at + timedelta(seconds=settings.SMS_CODE_TTL_SEC)
+            for sent_phone, sent_at in send_rows
+        )
+    }
+    return len(matched_users)
+
+
+def incident_details(
+    db: Session,
+    *,
+    incident: RiskIncident,
+    kind: str,
+    limit: int,
+    cursor: int,
+) -> dict:
+    if kind == "sms":
+        rows, total = _page_events(
+            db,
+            incident,
+            event_type=risk_repo.EVENT_SMS_SEND,
+            outcomes=("success",),
+            limit=limit,
+            cursor=cursor,
+        )
+        phone_values = {r.phone for r in rows if r.phone}
+        login_rows = list(
+            db.scalars(
+                select(BehaviorEvent)
+                .where(
+                    BehaviorEvent.event_type == risk_repo.EVENT_SMS_LOGIN,
+                    BehaviorEvent.subject_id == incident.subject_id,
+                    BehaviorEvent.phone.in_(phone_values),
+                    BehaviorEvent.outcome == "success",
+                    BehaviorEvent.occurred_at >= incident.window_start,
+                    BehaviorEvent.occurred_at
+                    < incident.window_end + timedelta(seconds=settings.SMS_CODE_TTL_SEC),
+                )
+                .order_by(BehaviorEvent.occurred_at.asc(), BehaviorEvent.id.asc())
+            ).all()
+        ) if phone_values else []
+        users = _user_map(db, {r.user_id for r in login_rows if r.user_id is not None})
+        items = []
+        for row in rows:
+            match = next(
+                (
+                    login
+                    for login in login_rows
+                    if login.phone == row.phone
+                    and login.occurred_at >= row.occurred_at
+                    and login.occurred_at
+                    <= row.occurred_at + timedelta(seconds=settings.SMS_CODE_TTL_SEC)
+                ),
+                None,
+            )
+            user = users.get(match.user_id) if match and match.user_id else None
+            items.append(
+                {
+                    "id": row.id,
+                    "occurred_at": row.occurred_at,
+                    "outcome": row.outcome,
+                    "phone": row.phone,
+                    "user_id": user.id if user else None,
+                    "username": user.username if user else None,
+                    "account_registered_at": user.created_at if user else None,
+                    "verified": match is not None,
+                }
+            )
+        distinct_accounts = _sms_distinct_accounts(db, incident)
+    elif kind == "oneclick":
+        rows, total = _page_events(
+            db,
+            incident,
+            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+            outcomes=("success", "failed"),
+            limit=limit,
+            cursor=cursor,
+        )
+        users = _user_map(db, {r.user_id for r in rows if r.user_id is not None})
+        items = []
+        for row in rows:
+            user = users.get(row.user_id) if row.user_id else None
+            items.append(
+                {
+                    "id": row.id,
+                    "occurred_at": row.occurred_at,
+                    "outcome": row.outcome,
+                    "phone": row.phone,
+                    "user_id": row.user_id,
+                    "username": user.username if user else None,
+                    "account_registered_at": user.created_at if user else None,
+                    "reason": row.reason,
+                }
+            )
+        distinct_accounts = _distinct_event_users(
+            db,
+            incident,
+            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+            outcomes=("success", "failed"),
+        )
+    else:
+        user_id = int(incident.subject_id)
+        filters = (
+            ComparisonRecord.user_id == user_id,
+            ComparisonRecord.created_at >= incident.window_start,
+            ComparisonRecord.created_at < incident.window_end,
+        )
+        total = int(db.scalar(select(func.count(ComparisonRecord.id)).where(*filters)) or 0)
+        rows = list(
+            db.scalars(
+                select(ComparisonRecord)
+                .where(*filters)
+                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
+                .offset(cursor)
+                .limit(limit)
+            ).all()
+        )
+        items = []
+        for row in rows:
+            prices = {
+                str(result.get("platform_name") or result.get("platform_id") or "未知平台"): (
+                    float(result["price"]) if result.get("price") is not None else None
+                )
+                for result in (row.comparison_results or [])
+            }
+            items.append(
+                {
+                    "id": row.id,
+                    "occurred_at": row.created_at,
+                    "outcome": row.status,
+                    "store_name": row.store_name,
+                    "product_names": row.product_names,
+                    "prices": prices,
+                    "saved_amount_yuan": (
+                        row.saved_amount_cents / 100
+                        if row.saved_amount_cents is not None
+                        else None
+                    ),
+                    "status": row.status,
+                }
+            )
+        distinct_accounts = None
+    next_cursor = cursor + len(items) if cursor + len(items) < total else None
+    return {
+        "kind": kind,
+        "incident_id": incident.id,
+        "event_count": incident.event_count,
+        "distinct_accounts": distinct_accounts,
+        "items": items,
+        "next_cursor": next_cursor,
+        "total": total,
+    }

app/admin/routers/ad_revenue.py

@@ -99,6 +99,7 @@ def get_ad_revenue_report(
         daily=[AdRevenueDaily(**d) for d in result["daily"]],
         hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
         type_stats={k: AdRevenueTypeStat(**v) for k, v in result["type_stats"].items()},
+        category_stats={k: AdRevenueTypeStat(**v) for k, v in result["category_stats"].items()},
         scene_stats={k: AdRevenueTypeStat(**v) for k, v in result["scene_stats"].items()},
         dau=result["dau"],
         total=result["total"],
@@ -107,6 +108,8 @@ def get_ad_revenue_report(
         total_revenue_yuan=result["total_revenue_yuan"],
         total_pangle_revenue_yuan=result["total_pangle_revenue_yuan"],
         total_pangle_api_revenue_yuan=result["total_pangle_api_revenue_yuan"],
+        pangle_api_revenue_complete=result["pangle_api_revenue_complete"],
+        pangle_latest_synced_at=result["pangle_latest_synced_at"],
         pangle_revenue_available=result["pangle_revenue_available"],
         total_expected_coin=result["total_expected_coin"],
         total_actual_coin=result["total_actual_coin"],

app/admin/routers/config.py

@@ -30,6 +30,12 @@ def _validate(key: str, value: Any) -> None:
     if t == "int":
         if not isinstance(value, int) or isinstance(value, bool) or value < 0:
             raise ValueError("需为非负整数")
+        minimum = CONFIG_DEFS[key].get("min")
+        maximum = CONFIG_DEFS[key].get("max")
+        if minimum is not None and value < minimum:
+            raise ValueError(f"不能小于 {minimum}")
+        if maximum is not None and value > maximum:
+            raise ValueError(f"不能大于 {maximum}")
     elif t == "int_list":
         if not isinstance(value, list) or not value:
             raise ValueError("需为非空整数列表")

app/admin/routers/risk_monitor.py

@@ -0,0 +1,331 @@
+"""风控监控：统计、风险事件列表/详情及忽略、封禁、解封。"""
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+
+from app.admin.audit import write_audit
+from app.admin.deps import (
+    AdminDb,
+    get_client_ip,
+    require_page,
+    require_role,
+)
+from app.admin.repositories import risk_monitor as repo
+from app.admin.schemas.risk_monitor import (
+    RestrictionRevokeResponse,
+    RiskActionRequest,
+    RiskActionResponse,
+    RiskDetailPage,
+    RiskIncidentPage,
+    RiskMonitorSummary,
+    RiskResetResponse,
+    RiskRuleConfig,
+)
+from app.core.config_schema import (
+    RISK_COMPARE_DAILY_THRESHOLD_KEY,
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+)
+from app.models.risk import RiskIncident, SubjectRestriction
+from app.repositories import app_config
+from app.repositories import risk as risk_repo
+
+router = APIRouter(
+    prefix="/admin/api/risk-monitor",
+    tags=["admin-risk-monitor"],
+    dependencies=[Depends(require_page("risk-monitor"))],
+)
+
+RiskKind = Literal["sms", "oneclick", "compare"]
+
+
+def _rule_config(db) -> RiskRuleConfig:
+    return RiskRuleConfig(
+        sms_hourly_threshold=risk_repo.get_rule_threshold(
+            db, risk_repo.RULE_SMS_HOURLY
+        ),
+        oneclick_daily_threshold=risk_repo.get_rule_threshold(
+            db, risk_repo.RULE_ONECLICK_DAILY
+        ),
+        compare_daily_threshold=risk_repo.get_rule_threshold(
+            db, risk_repo.RULE_COMPARE_DAILY
+        ),
+    )
+
+
+def _incident_or_404(db, incident_id: int, kind: RiskKind | None = None) -> RiskIncident:
+    incident = db.get(RiskIncident, incident_id)
+    if incident is None:
+        raise HTTPException(status_code=404, detail="风险事件不存在")
+    if kind and incident.rule_code != repo.KIND_TO_RULE[kind]:
+        raise HTTPException(status_code=404, detail="风险事件类型不匹配")
+    return incident
+
+
+@router.get("/summary", response_model=RiskMonitorSummary, summary="风控监控顶部卡片")
+def get_summary(db: AdminDb) -> RiskMonitorSummary:
+    return RiskMonitorSummary(**repo.summary(db))
+
+
+@router.get("/rules", response_model=RiskRuleConfig, summary="读取风控规则阈值")
+def get_rules(db: AdminDb) -> RiskRuleConfig:
+    return _rule_config(db)
+
+
+@router.patch("/rules", response_model=RiskRuleConfig, summary="修改风控规则并即时重算")
+def update_rules(
+    body: RiskRuleConfig,
+    request: Request,
+    admin: Annotated[object, Depends(require_role("operator", "tech"))],
+    db: AdminDb,
+) -> RiskRuleConfig:
+    before = _rule_config(db).model_dump()
+    after = body.model_dump()
+    values = (
+        (RISK_SMS_HOURLY_THRESHOLD_KEY, body.sms_hourly_threshold),
+        (RISK_ONECLICK_DAILY_THRESHOLD_KEY, body.oneclick_daily_threshold),
+        (RISK_COMPARE_DAILY_THRESHOLD_KEY, body.compare_daily_threshold),
+    )
+    for key, value in values:
+        app_config.set_value(
+            db, key, value, admin_id=admin.id, commit=False
+        )
+
+    now = risk_repo.utcnow()
+    risk_repo.reconcile_behavior_rule(
+        db, rule_code=risk_repo.RULE_SMS_HOURLY, at=now, commit=False
+    )
+    risk_repo.reconcile_behavior_rule(
+        db, rule_code=risk_repo.RULE_ONECLICK_DAILY, at=now, commit=False
+    )
+    risk_repo.reconcile_compare_rule(db, at=now, commit=False)
+    write_audit(
+        db,
+        admin,
+        action="risk.rules.update",
+        target_type="risk_rule",
+        target_id="all",
+        detail={"before": before, "after": after},
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return _rule_config(db)
+
+
+@router.post(
+    "/reset",
+    response_model=RiskResetResponse,
+    summary="重置三类待处理报警并从零重新累计",
+)
+def reset_alerts(
+    request: Request,
+    admin: Annotated[object, Depends(require_role("operator", "tech"))],
+    db: AdminDb,
+) -> RiskResetResponse:
+    reset_at = risk_repo.utcnow()
+    counts = repo.reset_open_alerts(db, admin_id=admin.id, reset_at=reset_at)
+    total = sum(counts.values())
+    write_audit(
+        db,
+        admin,
+        action="risk.alerts.reset",
+        target_type="risk_monitor",
+        target_id="all",
+        detail={
+            "reset_at": reset_at.isoformat(),
+            "reset_incident_count": total,
+            "reset_counts": counts,
+            "restrictions_changed": False,
+        },
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return RiskResetResponse(
+        reset_at=reset_at,
+        reset_incident_count=total,
+        reset_counts=counts,
+    )
+
+
+@router.get(
+    "/incidents/{kind}",
+    response_model=RiskIncidentPage,
+    summary="按短信/一键登录/比价读取风险事件",
+)
+def get_incidents(
+    kind: RiskKind,
+    db: AdminDb,
+    incident_status: Annotated[
+        Literal["open", "ignored", "blocked", "resolved", "all"], Query(alias="status")
+    ] = "open",
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+    cursor: Annotated[int, Query(ge=0)] = 0,
+) -> RiskIncidentPage:
+    if kind == "compare":
+        repo.sync_today_compare_incidents(db)
+    items, next_cursor, total = repo.list_incidents(
+        db, kind=kind, status=incident_status, limit=limit, cursor=cursor
+    )
+    return RiskIncidentPage(items=items, next_cursor=next_cursor, total=total)
+
+
+@router.get(
+    "/incidents/{kind}/{incident_id}/details",
+    response_model=RiskDetailPage,
+    summary="风险事件当期事实明细",
+)
+def get_incident_details(
+    kind: RiskKind,
+    incident_id: int,
+    db: AdminDb,
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+    cursor: Annotated[int, Query(ge=0)] = 0,
+) -> RiskDetailPage:
+    incident = _incident_or_404(db, incident_id, kind)
+    return RiskDetailPage(
+        **repo.incident_details(
+            db, incident=incident, kind=kind, limit=limit, cursor=cursor
+        )
+    )
+
+
+@router.post(
+    "/incidents/{incident_id}/ignore",
+    response_model=RiskActionResponse,
+    summary="忽略风险事件",
+)
+def ignore_incident(
+    incident_id: int,
+    body: RiskActionRequest,
+    request: Request,
+    admin: Annotated[object, Depends(require_role("operator", "tech"))],
+    db: AdminDb,
+) -> RiskActionResponse:
+    incident = _incident_or_404(db, incident_id)
+    if incident.status != "open":
+        raise HTTPException(status_code=409, detail="该风险事件已处理")
+    incident.status = "ignored"
+    incident.action_reason = body.reason.strip() or "管理员确认忽略"
+    incident.handled_by = admin.id
+    incident.handled_at = risk_repo.utcnow()
+    write_audit(
+        db,
+        admin,
+        action="risk.incident.ignore",
+        target_type="risk_incident",
+        target_id=incident.id,
+        detail={
+            "rule_code": incident.rule_code,
+            "subject_type": incident.subject_type,
+            "subject_id": incident.subject_id,
+            "reason": incident.action_reason,
+        },
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return RiskActionResponse(incident_id=incident.id, status=incident.status)
+
+
+@router.post(
+    "/incidents/{incident_id}/block",
+    response_model=RiskActionResponse,
+    summary="封禁风险主体",
+)
+def block_incident(
+    incident_id: int,
+    body: RiskActionRequest,
+    request: Request,
+    admin: Annotated[object, Depends(require_role("operator", "tech"))],
+    db: AdminDb,
+) -> RiskActionResponse:
+    incident = _incident_or_404(db, incident_id)
+    if incident.status not in ("open", "blocked"):
+        raise HTTPException(status_code=409, detail="该风险事件已处理")
+    if incident.subject_type == "device":
+        if incident.subject_id.startswith("legacy-ip:"):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="旧客户端未上报设备ID，不能执行设备封禁",
+            )
+        scope = risk_repo.SCOPE_AUTH_DEVICE
+    elif incident.subject_type == "user":
+        scope = risk_repo.SCOPE_ECONOMIC_ACCOUNT
+    else:  # pragma: no cover - 当前三条规则只会生成 device/user
+        raise HTTPException(status_code=400, detail="该主体类型暂不支持封禁")
+
+    reason = body.reason.strip() or "风控规则命中"
+    restriction = risk_repo.activate_restriction(
+        db,
+        subject_type=incident.subject_type,
+        subject_id=incident.subject_id,
+        scope=scope,
+        reason=reason,
+        incident_id=incident.id,
+        admin_id=admin.id,
+    )
+    incident.status = "blocked"
+    incident.action_reason = reason
+    incident.handled_by = admin.id
+    incident.handled_at = risk_repo.utcnow()
+    write_audit(
+        db,
+        admin,
+        action="risk.subject.block",
+        target_type=incident.subject_type,
+        target_id=incident.subject_id,
+        detail={
+            "incident_id": incident.id,
+            "rule_code": incident.rule_code,
+            "scope": scope,
+            "reason": reason,
+        },
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return RiskActionResponse(
+        incident_id=incident.id,
+        status=incident.status,
+        restriction_id=restriction.id,
+    )
+
+
+@router.post(
+    "/restrictions/{restriction_id}/revoke",
+    response_model=RestrictionRevokeResponse,
+    summary="解除主体限制",
+)
+def revoke_restriction(
+    restriction_id: int,
+    request: Request,
+    admin: Annotated[object, Depends(require_role("operator", "tech"))],
+    db: AdminDb,
+) -> RestrictionRevokeResponse:
+    restriction = db.get(SubjectRestriction, restriction_id)
+    if restriction is None:
+        raise HTTPException(status_code=404, detail="限制记录不存在")
+    if restriction.active:
+        risk_repo.revoke_restriction(db, restriction=restriction, admin_id=admin.id)
+        if restriction.incident_id:
+            incident = db.get(RiskIncident, restriction.incident_id)
+            if incident and incident.status == "blocked":
+                incident.status = "resolved"
+                incident.handled_by = admin.id
+                incident.handled_at = risk_repo.utcnow()
+        write_audit(
+            db,
+            admin,
+            action="risk.subject.unblock",
+            target_type=restriction.subject_type,
+            target_id=restriction.subject_id,
+            detail={"restriction_id": restriction.id, "scope": restriction.scope},
+            ip=get_client_ip(request),
+            commit=False,
+        )
+        db.commit()
+    return RestrictionRevokeResponse(restriction_id=restriction.id)

app/admin/schemas/ad_revenue.py

@@ -40,6 +40,8 @@ class AdRevenueRecord(BaseModel):
     expected_coin: int = Field(..., description="按公式复算应发金币")
     actual_coin: int = Field(..., description="实际入账金币")
     matched: bool = Field(..., description="复算与实发是否一致")
+    adn: str | None = Field(None, description="本条发奖对应的实际填充 ADN 子渠道")
+    slot_id: str | None = Field(None, description="本条发奖对应的底层 mediation rit")
 
 
 class AdRevenueDaily(BaseModel):
@@ -47,12 +49,12 @@ class AdRevenueDaily(BaseModel):
 
     date: str = Field(..., description="北京时间 YYYY-MM-DD")
     impressions: int = Field(..., description="当天展示条数合计")
-    revenue_yuan: float = Field(..., description="当天客户端有效预估收益合计(元;eCPM 折算)")
+    revenue_yuan: float = Field(..., description="当天客户端 SDK 展示预估合计(元;后端留存 eCPM 折算)")
     pangle_revenue_yuan: float | None = Field(
-        None, description="当天穿山甲后台预估收益(元;GroMore revenue);非全量视图/无数据为空"
+        None, description="当天 GroMore 排序价预估(元;revenue,非结算收入);非全量视图/无数据为空"
     )
     pangle_api_revenue_yuan: float | None = Field(
-        None, description="当天穿山甲收益Api(元;GroMore api_revenue,更接近结算);未配/当天/无数据为空"
+        None, description="当天 ADN Reporting API 收益(元;GroMore api_revenue);未配/当天/无数据为空"
     )
     expected_coin: int = Field(..., description="当天应发金币合计")
     actual_coin: int = Field(..., description="当天实发金币合计")
@@ -69,10 +71,11 @@ class AdRevenueHourly(BaseModel):
 
 
 class AdRevenueTypeStat(BaseModel):
-    """按广告类型(ad_type)的小计:展示条数 + 预估收益(eCPM 由前端用 收益÷展示×1000 算)。"""
+    """展示条数、SDK 展示预估收益与按展示次数加权的 SDK eCPM。"""
 
     impressions: int = Field(..., description="该类型展示条数合计")
     revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
+    ecpm_yuan: float = Field(..., description="按展示次数加权的 SDK eCPM(元/千次)")
 
 
 class AdRevenueRow(BaseModel):
@@ -98,15 +101,15 @@ class AdRevenueRow(BaseModel):
     ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
     revenue_yuan: float = Field(
         ...,
-        description="本次有效展示预估收益(元)= eCPM元 ÷ 1000;纯发奖、激励视频提前关闭/时长不足=0",
+        description="本次 SDK 展示预估收益(元)=后端留存 eCPM 元 ÷ 1000;是否满足发奖条件不改变展示收入预估",
     )
     row_revenue_yuan: float | None = Field(
         None,
         description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
         "其它行为空(前端回退取 revenue_yuan)。不进合计/趋势,避免与展示侧重复计",
     )
-    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空")
-    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
+    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);历史或未上报展示来源为空")
+    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);历史或未上报展示来源为空")
     # ── 发奖侧 ──
     has_reward: bool = Field(..., description="是否有发奖记录(激励视频合并行 / 信息流整场发奖行=True;纯展示=False)")
     status: str | None = Field(None, description="发奖状态 granted/closed_early/too_short/…;纯展示为空")
@@ -140,7 +143,11 @@ class AdRevenueReportOut(BaseModel):
     )
     type_stats: dict[str, AdRevenueTypeStat] = Field(
         default_factory=dict,
-        description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
+        description="按原始广告类型(ad_type)小计，供筛选与排查使用",
+    )
+    category_stats: dict[str, AdRevenueTypeStat] = Field(
+        default_factory=dict,
+        description="按经营分类小计：draw=draw+历史 feed，video=reward_video+withdrawal_video",
     )
     scene_stats: dict[str, AdRevenueTypeStat] = Field(
         default_factory=dict,
@@ -156,20 +163,28 @@ class AdRevenueReportOut(BaseModel):
     total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
     truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
     total_impressions: int = Field(..., description="全量展示条数合计")
-    total_revenue_yuan: float = Field(..., description="全量客户端有效预估收益合计(元;eCPM 折算)")
+    total_revenue_yuan: float = Field(..., description="全量客户端 SDK 展示预估合计(元;后端留存 eCPM 折算)")
     total_pangle_revenue_yuan: float | None = Field(
         None,
-        description="全量穿山甲后台预估收益合计(元;GroMore revenue)。穿山甲无用户/类型/场景维度,"
+        description="全量 GroMore 排序价预估合计(元;revenue,非结算收入)。GroMore 无用户/类型/场景维度,"
         "仅「全量视图」(未按 user_id/ad_type/feed_scene 过滤)时有值,否则为 null",
     )
     total_pangle_api_revenue_yuan: float | None = Field(
         None,
-        description="全量穿山甲收益Api合计(元;GroMore api_revenue,各 ADN 回传、更接近结算);"
+        description="全量 ADN Reporting API 收益合计(元;GroMore api_revenue,仅已配置回传的 ADN);"
         "未配 Reporting / 查当天 / 非全量视图 时为 null",
     )
+    pangle_api_revenue_complete: bool = Field(
+        False,
+        description="所选每一天是否都已在 D+1 14:00 后完成 API 同步窗口；不代表覆盖全部 ADN 或最终结算",
+    )
+    pangle_latest_synced_at: datetime | None = Field(
+        None,
+        description="所选范围穿山甲/GroMore 日报最近同步时间",
+    )
     pangle_revenue_available: bool = Field(
         False,
-        description="本次结果是否带穿山甲后台收益(=全量视图且已同步到数据)。false 时前端「穿山甲收益」显示「-」",
+        description="本次结果是否带 GroMore/ADN 收益(=全量视图且已同步到数据)。false 时前端显示「-」",
     )
     total_expected_coin: int = Field(..., description="全量应发金币合计")
     total_actual_coin: int = Field(..., description="全量实发金币合计")

app/admin/schemas/risk_monitor.py

@@ -0,0 +1,104 @@
+"""风控监控后台接口 schema。"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class RiskSummaryCard(BaseModel):
+    kind: str
+    alert_subject_count: int
+    today_total: int
+    threshold: int
+    rule_text: str
+
+
+class RiskMonitorSummary(BaseModel):
+    date: str
+    updated_at: datetime
+    cards: list[RiskSummaryCard]
+
+
+class RiskRuleConfig(BaseModel):
+    sms_hourly_threshold: int = Field(ge=1, le=5)
+    oneclick_daily_threshold: int = Field(ge=1, le=100_000)
+    compare_daily_threshold: int = Field(ge=1, le=100)
+
+
+class RiskIncidentItem(BaseModel):
+    incident_id: int
+    kind: str
+    subject_type: str
+    subject_id: str
+    device_model: str | None = None
+    first_used_at: datetime | None = None
+    phone: str | None = None
+    user_id: int | None = None
+    registered_at: datetime | None = None
+    common_device_id: str | None = None
+    window_start: datetime
+    window_end: datetime
+    first_event_at: datetime
+    triggered_at: datetime
+    last_event_at: datetime
+    event_count: int
+    status: str
+    restricted: bool = False
+    restriction_id: int | None = None
+
+
+class RiskIncidentPage(BaseModel):
+    items: list[RiskIncidentItem]
+    next_cursor: int | None = None
+    total: int
+
+
+class RiskDetailItem(BaseModel):
+    id: int
+    occurred_at: datetime
+    outcome: str
+    phone: str | None = None
+    user_id: int | None = None
+    username: str | None = None
+    account_registered_at: datetime | None = None
+    verified: bool | None = None
+    reason: str | None = None
+    store_name: str | None = None
+    product_names: str | None = None
+    prices: dict[str, float | None] | None = None
+    saved_amount_yuan: float | None = None
+    status: str | None = None
+
+
+class RiskDetailPage(BaseModel):
+    kind: str
+    incident_id: int
+    event_count: int
+    distinct_accounts: int | None = None
+    items: list[RiskDetailItem]
+    next_cursor: int | None = None
+    total: int
+
+
+class RiskActionRequest(BaseModel):
+    reason: str = Field("", max_length=256)
+
+
+class RiskActionResponse(BaseModel):
+    ok: bool = True
+    incident_id: int
+    status: str
+    restriction_id: int | None = None
+
+
+class RiskResetResponse(BaseModel):
+    ok: bool = True
+    reset_at: datetime
+    reset_incident_count: int
+    reset_counts: dict[str, int]
+
+
+class RestrictionRevokeResponse(BaseModel):
+    ok: bool = True
+    restriction_id: int

app/api/v1/ad.py

@@ -289,7 +289,8 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     """客户端在广告展示后(onAdShow 读 getShowEcpm)上报 eCPM,落库做内部收益统计/对账。
 
     Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
-    丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
+    丢一两条不影响发奖业务(收入另由 ADN Reporting API 对账)。eCPM 与发奖(S2S)是两条独立流,
+    不逐条关联。
     """
     attributed_trace_id = crud_ecpm.attributable_trace_id(
         db,

app/api/v1/auth.py

@@ -37,6 +37,7 @@ from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_ph
 from app.integrations.sms import SmsError, send_code, verify_code
 from app.repositories import onboarding as onboarding_repo
 from app.repositories import phone_rebind as rebind_repo
+from app.repositories import risk as risk_repo
 from app.repositories import user as user_repo
 from app.schemas.auth import (
     JverifyLoginRequest,
@@ -70,6 +71,18 @@ SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 5     # 每小时上限
 SMS_SEND_MAX_PER_DAY_PER_DEVICE = 20     # 每天上限(再叠一层日封顶,挡低频长时间轰炸)
 
 
+def _client_ip(request: Request) -> str:
+    xff = request.headers.get("x-forwarded-for")
+    if xff:
+        return xff.split(",")[0].strip()
+    return request.client.host if request.client else ""
+
+
+def _device_subject(device_id: str, request: Request) -> str:
+    # 兼容旧客户端空 device_id；新客户端均上传稳定硬件标识。
+    return device_id or f"legacy-ip:{_client_ip(request)}"
+
+
 def _login_response(
     user, *, onboarding_completed: bool, force_onboarding: bool = False
 ) -> TokenWithUser:
@@ -87,7 +100,15 @@ def _login_response(
 # ===================== 极光一键登录 =====================
 
 @router.post("/jverify-login", response_model=TokenWithUser, summary="极光一键登录")
-def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
+def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
+    subject_id = _device_subject(req.device_id, request)
+    if req.device_id and risk_repo.is_restricted(
+        db,
+        subject_type="device",
+        subject_id=req.device_id,
+        scope=risk_repo.SCOPE_AUTH_DEVICE,
+    ):
+        raise HTTPException(status_code=403, detail="当前设备环境异常，暂无法登录")
     logger.info(
         "jverify_login operator=%s token_len=%d",
         req.operator or "-",
@@ -97,10 +118,37 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     try:
         phone = verify_and_get_phone(req.login_token)
     except JiguangError as e:
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+            subject_type="device",
+            subject_id=subject_id,
+            device_id=req.device_id or None,
+            device_model=req.device_model or None,
+            client_ip=_client_ip(request),
+            outcome="failed",
+            reason=str(e),
+            details={"operator": req.operator or None},
+            evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
+        )
         logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
         raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
 
     user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="jverify")
+    risk_repo.record_behavior_event(
+        db,
+        event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+        subject_type="device",
+        subject_id=subject_id,
+        user_id=user.id,
+        device_id=req.device_id or None,
+        device_model=req.device_model or None,
+        phone=phone,
+        client_ip=_client_ip(request),
+        outcome="success",
+        details={"operator": req.operator or None},
+        evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
+    )
     if user.status != "active":
         raise HTTPException(status_code=403, detail="account disabled")
 
@@ -116,12 +164,34 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     response_model=SmsSendResponse,
     summary="发送短信验证码",
 )
-def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
+def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendResponse:
+    subject_id = _device_subject(req.device_id, request)
+    if req.device_id and risk_repo.is_restricted(
+        db,
+        subject_type="device",
+        subject_id=req.device_id,
+        scope=risk_repo.SCOPE_AUTH_DEVICE,
+    ):
+        raise HTTPException(status_code=403, detail="当前设备环境异常，暂无法发送验证码")
     # 测试账号:不真发短信(号码非真实手机号,真发会失败/浪费),直接放行让客户端进入填码界面。
     # 真正的"免验证码"在 /sms/login 跳过校验;每日上限也在 login 处算(send 不耗额度)。
     # (也不受下面设备发码限流约束:QA 联调要反复发码。)
     if test_account.is_test_account(req.phone):
         logger.info("test_account sms_send short-circuit (不真发)")
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_SEND,
+            subject_type="device",
+            subject_id=subject_id,
+            device_id=req.device_id or None,
+            device_model=req.device_model or None,
+            phone=req.phone,
+            client_ip=_client_ip(request),
+            # 测试号不真发短信，也豁免正式频控；保留流水供排查，但不能污染
+            # “今日短信下发”与短信风控报警。
+            outcome="test",
+            details={"mock": True, "test_account": True},
+        )
         return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
 
     # 发码防刷:同一设备(device_id) + 同一 IP,每小时 / 每天两道闸,**均只按成功发码计数**。
@@ -139,6 +209,18 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
     try:
         cooldown = send_code(req.phone)
     except SmsError as e:
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_SEND,
+            subject_type="device",
+            subject_id=subject_id,
+            device_id=req.device_id or None,
+            device_model=req.device_model or None,
+            phone=req.phone,
+            client_ip=_client_ip(request),
+            outcome="failed",
+            reason=str(e),
+        )
         raise HTTPException(status_code=e.status_code, detail=str(e)) from e
 
     # 发码成功 → 两道闸各 +1(被单号冷却挡下的重发走不到这里,故不占额度)
@@ -146,6 +228,20 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
 
     from app.core.config import settings  # 局部 import 避免循环
 
+    risk_repo.record_behavior_event(
+        db,
+        event_type=risk_repo.EVENT_SMS_SEND,
+        subject_type="device",
+        subject_id=subject_id,
+        device_id=req.device_id or None,
+        device_model=req.device_model or None,
+        phone=req.phone,
+        client_ip=_client_ip(request),
+        outcome="success",
+        details={"mock": settings.SMS_MOCK},
+        evaluate_rule=risk_repo.RULE_SMS_HOURLY,
+    )
+
     return SmsSendResponse(sent=True, mock=settings.SMS_MOCK, cooldown_sec=cooldown)
 
 
@@ -155,6 +251,7 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
     summary="手机号+验证码登录",
 )
 def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
+    subject_id = _device_subject(req.device_id, request)
     # 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。
     # 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
     # 测试账号走自己的每日额度、不受下面 (设备+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。
@@ -162,6 +259,19 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
         if not test_account.try_consume_quota():
             raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
         user = user_repo.upsert_user_for_login(db, phone=req.phone, register_channel="sms")
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_LOGIN,
+            subject_type="device",
+            subject_id=subject_id,
+            user_id=user.id,
+            device_id=req.device_id or None,
+            device_model=req.device_model or None,
+            phone=req.phone,
+            client_ip=_client_ip(request),
+            outcome="success",
+            details={"test_account": True},
+        )
         if user.status != "active":
             raise HTTPException(status_code=403, detail="account disabled")
         logger.info("sms_login test_account ok user_id=%d phone=%s", user.id, mask_phone(req.phone))
@@ -183,9 +293,33 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
     )
 
     if not verify_code(req.phone, req.code):
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_LOGIN,
+            subject_type="device",
+            subject_id=subject_id,
+            device_id=req.device_id or None,
+            device_model=req.device_model or None,
+            phone=req.phone,
+            client_ip=_client_ip(request),
+            outcome="failed",
+            reason="invalid sms code",
+        )
         raise HTTPException(status_code=400, detail="invalid sms code")
 
     user = user_repo.upsert_user_for_login(db, phone=req.phone, register_channel="sms")
+    risk_repo.record_behavior_event(
+        db,
+        event_type=risk_repo.EVENT_SMS_LOGIN,
+        subject_type="device",
+        subject_id=subject_id,
+        user_id=user.id,
+        device_id=req.device_id or None,
+        device_model=req.device_model or None,
+        phone=req.phone,
+        client_ip=_client_ip(request),
+        outcome="success",
+    )
     if user.status != "active":
         raise HTTPException(status_code=403, detail="account disabled")
 

app/api/v1/compare.py

@@ -28,19 +28,30 @@ import httpx
 from fastapi import APIRouter, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
-from app.api.deps import OptionalUser
+from app.api.deps import DbSession, OptionalUser
 from app.core.config import settings
 from app.core.logging import trace_id_ctx
 from app.core.pricebot_client import get_pricebot_client
 from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import comparison as crud_compare
+from app.repositories import risk as risk_repo
 
 logger = logging.getLogger("shagua.compare")
 
 router = APIRouter(prefix="/api/v1", tags=["compare"])
 
 
+def _ensure_compare_allowed(user, db) -> None:
+    if user and risk_repo.is_restricted(
+        db,
+        subject_type="user",
+        subject_id=str(user.id),
+        scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+    ):
+        raise HTTPException(status_code=403, detail="账号存在异常，该功能暂不可用")
+
+
 # ============================================================
 # harvest 落库(阻塞 SQLAlchemy → run_in_threadpool,独立 SessionLocal,不阻塞事件循环;
 # 任何写库异常都吞掉、绝不连累比价透传返回 —— 同 coupon.py 现有 best-effort 写法)。
@@ -52,10 +63,14 @@ def _harvest_running_blocking(
     device_id: str | None, device_info: dict | None, trace_url: str | None,
 ) -> None:
     with SessionLocal() as db:
-        crud_compare.harvest_running(
+        rec = crud_compare.harvest_running(
             db, trace_id=trace_id, user_id=user_id, business_type=business_type,
             device_id=device_id, device_info=device_info, trace_url=trace_url,
         )
+        if rec.user_id is not None:
+            risk_repo.sync_compare_incident(
+                db, user_id=rec.user_id, at=rec.created_at
+            )
     logger.info(
         "harvest running row (user=%s)", user_id,
         extra={"phase": "harvest_running", "status": "running", "user_id": user_id},
@@ -72,6 +87,10 @@ def _harvest_done_blocking(
             business_type=business_type, device_id=device_id,
             device_info=device_info, trace_url=trace_url,
         )
+        if rec.user_id is not None:
+            risk_repo.sync_compare_incident(
+                db, user_id=rec.user_id, at=rec.created_at
+            )
         logger.info(
             "harvest done → %s saved=%s newly=%s", rec.status,
             rec.saved_amount_cents, newly_success,
@@ -199,25 +218,33 @@ async def _forward(
 
 
 @router.post("/intent/recognize", summary="外卖比价 Phase 1 意图识别 (透传 + 建 running 行)")
-async def intent_recognize(request: Request, user: OptionalUser) -> dict[str, Any]:
+async def intent_recognize(
+    request: Request, user: OptionalUser, db: DbSession
+) -> dict[str, Any]:
+    _ensure_compare_allowed(user, db)
     resp, _, _ = await _forward(request, "/api/intent/recognize", user)
     return resp
 
 
 @router.post("/intent/step", summary="外卖比价 Phase 1 多帧意图识别 (透传, 仅淘宝源)")
-async def intent_step(request: Request, user: OptionalUser) -> dict[str, Any]:
+async def intent_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
+    _ensure_compare_allowed(user, db)
     resp, _, _ = await _forward(request, "/api/intent/step", user)
     return resp
 
 
 @router.post("/intent/precoupon/step", summary="外卖比价 Phase 0 识别前先用券 (透传, 仅美团源)")
-async def intent_precoupon_step(request: Request, user: OptionalUser) -> dict[str, Any]:
+async def intent_precoupon_step(
+    request: Request, user: OptionalUser, db: DbSession
+) -> dict[str, Any]:
+    _ensure_compare_allowed(user, db)
     resp, _, _ = await _forward(request, "/api/intent/precoupon/step", user)
     return resp
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(request: Request, user: OptionalUser) -> dict[str, Any]:
+async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
+    _ensure_compare_allowed(user, db)
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
     # (单平台中途 done 被 pricebot 改写成 wait+continue=true,不会命中这里,同 coupon 语义。)
@@ -237,7 +264,10 @@ async def price_step(request: Request, user: OptionalUser) -> dict[str, Any]:
 
 
 @router.post("/trace/epilogue", summary="比价结果页尾声帧 (透传到 pricebot, 用户视角截图入 trace)")
-async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]:
+async def trace_epilogue(
+    request: Request, user: OptionalUser, db: DbSession
+) -> dict[str, Any]:
+    _ensure_compare_allowed(user, db)
     # App 收到 done、渲染完结果页后,把自己页面的截图(base64)传给 pricebot 存进 trace
     # 目录并触发重传 —— trace 里补上"用户实际看到的汇总页"(步骤帧只有目标 App 画面)。
     # body ~几百 KB(截图 base64), 纯透传壳(harvest_first_frame=False:不建行/不落库,
@@ -250,7 +280,10 @@ async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]
 
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
-async def trace_finalize(request: Request, user: OptionalUser) -> dict[str, Any]:
+async def trace_finalize(
+    request: Request, user: OptionalUser, db: DbSession
+) -> dict[str, Any]:
+    _ensure_compare_allowed(user, db)
     # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
     # app-server 顺手把该 trace 的 running 行更新成夭折终态(**不降级 success**)。
     # 客户端 finalize body 带 status(cancelled/failed)+ reason;老客户端只带 trace_id →

app/api/v1/compare_record.py

@@ -19,6 +19,7 @@ from app.api.deps import CurrentUser, DbSession
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
+from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
     CompareStartReserveIn,
     CompareStartReserveOut,
@@ -47,8 +48,15 @@ def reserve_compare_start(
     user: CurrentUser,
     db: DbSession,
 ) -> CompareStartReserveOut:
+    if risk_repo.is_restricted(
+        db,
+        subject_type="user",
+        subject_id=str(user.id),
+        scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+    ):
+        raise HTTPException(status_code=403, detail="账号存在异常，该功能暂不可用")
     try:
-        _, used = crud_compare.reserve_daily_start(
+        rec, used = crud_compare.reserve_daily_start(
             db,
             user_id=user.id,
             trace_id=payload.trace_id,
@@ -65,6 +73,8 @@ def reserve_compare_start(
             status_code=status.HTTP_409_CONFLICT,
             detail="比价任务标识冲突，请重新发起",
         ) from None
+    # 风控阈值由后台动态配置，不能再只在固定的 100 次业务上限处同步。
+    risk_repo.sync_compare_incident(db, user_id=user.id, at=rec.created_at)
     return CompareStartReserveOut(
         limit=crud_compare.DAILY_COMPARE_START_LIMIT,
         used=used,
@@ -83,6 +93,13 @@ def report_record(
     db: DbSession,
     background_tasks: BackgroundTasks,
 ) -> ComparisonRecordCreatedOut:
+    if risk_repo.is_restricted(
+        db,
+        subject_type="user",
+        subject_id=str(user.id),
+        scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+    ):
+        raise HTTPException(status_code=403, detail="账号存在异常，该功能暂不可用")
     rec = crud_compare.upsert_record(db, user_id=user.id, payload=payload)
     # LLM 调用明细异步回填:同机拉 pricebot llm_calls 最长 5s 且是 best-effort,放后台
     # 任务做,不阻塞上报响应(顺带给 pricebot 落盘留足余量)。upsert 已 commit,后台用

app/api/v1/tasks.py

@@ -11,6 +11,7 @@ import logging
 from fastapi import APIRouter, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.repositories import risk as risk_repo
 from app.repositories import task as crud_task
 from app.schemas.welfare import TaskClaimResultOut, TaskListOut, TaskOut
 
@@ -36,6 +37,13 @@ def list_tasks(user: CurrentUser, db: DbSession) -> TaskListOut:
     summary="领取任务奖励",
 )
 def claim_task(task_key: str, user: CurrentUser, db: DbSession) -> TaskClaimResultOut:
+    if risk_repo.is_restricted(
+        db,
+        subject_type="user",
+        subject_id=str(user.id),
+        scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+    ):
+        raise HTTPException(status_code=403, detail="账号存在异常，该功能暂不可用")
     try:
         coin, balance = crud_task.claim_task(db, user.id, task_key)
     except crud_task.UnknownTaskError as e:

app/api/v1/wallet.py

@@ -20,8 +20,9 @@ from app.core.rewards import (
     WITHDRAW_MAX_CENTS,
     WITHDRAW_MIN_CENTS,
 )
-from app.repositories import wallet as crud_wallet
 from app.models.user import User
+from app.repositories import risk as risk_repo
+from app.repositories import wallet as crud_wallet
 from app.schemas.welfare import (
     BindWechatRequest,
     BindWechatResultOut,
@@ -204,6 +205,13 @@ def withdraw_info(
     dependencies=[Depends(rate_limit(5, 60, "withdraw"))],  # IP 级粗限流;用户级未完成单限制在仓库层
 )
 def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut:
+    if risk_repo.is_restricted(
+        db,
+        subject_type="user",
+        subject_id=str(user.id),
+        scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+    ):
+        raise HTTPException(status_code=403, detail="账号存在异常，该功能暂不可用")
     # 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦
     if not settings.wxpay_configured:
         raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="wechat pay not configured")
@@ -222,11 +230,6 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw
         ) from e
     except crud_wallet.WechatNotBoundError as 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.WithdrawTierUnavailableError as e:
         # 福利页档位闸(7-9):次数满/已选其他额度。正常客户端已按 tiers 预拦,此处兜底防绕过。
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="今日额度已达上限") from e

app/core/config.py

@@ -318,7 +318,7 @@ class Settings(BaseSettings):
     # ===== 穿山甲 GroMore 数据 API(报表收益拉取,T+1)=====
     # ⚠️ 与上面发奖回调的 m-key 是【两套完全不同的凭证】:这三样在穿山甲后台
     # 「接入中心 → GroMore-API → 聚合数据报告 API」文档页领取(user_id / role_id / Security Key),
-    # 仅用于按天拉 GroMore 收益报表(revenue 预估收益 + api_revenue 收益Api),不参与发奖。
+    # 仅用于按天拉 GroMore 报表(revenue 排序价预估 + api_revenue ADN Reporting 收益),不参与发奖。
     # 该 API 只能查【GroMore 聚合代码位】的数据(=我们 useMediation 的口径),非穿山甲 SDK 数据;
     # 且不提供用户/设备维度(官方明确),故收益只能落到 日期×代码位 汇总,不能挂到逐条事件。
     # 子账号(role_id≠user_id)需主账号在「角色管理」授予「查看全部数据」权限,否则查不到

app/core/config_schema.py

@@ -11,6 +11,10 @@ from typing import Any
 
 from app.core import rewards as r
 
+RISK_SMS_HOURLY_THRESHOLD_KEY = "risk_sms_hourly_threshold"
+RISK_ONECLICK_DAILY_THRESHOLD_KEY = "risk_oneclick_daily_threshold"
+RISK_COMPARE_DAILY_THRESHOLD_KEY = "risk_compare_daily_threshold"
+
 # type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int / bool / enum
 # hidden=True:仍是合法可配项(业务照常 get_value / admin 可经专用端点读写),但**不在通用
 #   「系统配置」页渲染**(admin/routers/config.py:list_config 按此过滤)。用于把已下线/已改由
@@ -106,4 +110,36 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
             "default 兜底未登记的模型。改价只影响之后回填的新记录,历史记录用当时价格快照。"
         ),
     },
+    # 风控监控页专用配置。复用 app_config 持久化，但不在通用「系统配置」页重复展示；
+    # 由 /admin/api/risk-monitor/rules 专用接口整组校验、审计和即时重算当前窗口。
+    RISK_SMS_HOURLY_THRESHOLD_KEY: {
+        "default": 5,
+        "label": "短信设备每小时告警阈值",
+        "group": "风控",
+        "type": "int",
+        "min": 1,
+        "max": 5,
+        "hidden": True,
+        "help": "同一设备在北京时间同一自然小时内成功下发短信达到该次数时告警；不得高于现有每小时 5 次的发送上限。",
+    },
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY: {
+        "default": 20,
+        "label": "一键登录设备每日告警阈值",
+        "group": "风控",
+        "type": "int",
+        "min": 1,
+        "max": 100_000,
+        "hidden": True,
+        "help": "同一设备在北京时间同一自然日内一键登录成功和失败合计达到该次数时告警。",
+    },
+    RISK_COMPARE_DAILY_THRESHOLD_KEY: {
+        "default": 100,
+        "label": "比价账户每日告警阈值",
+        "group": "风控",
+        "type": "int",
+        "min": 1,
+        "max": 100,
+        "hidden": True,
+        "help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
+    },
 }

app/integrations/pangle_report.py

@@ -14,8 +14,8 @@
   - 只返回【GroMore 聚合代码位】在 GroMore 内的数据(=我们 useMediation 的口径),
     查不到穿山甲 SDK 自身的数据;
   - **不提供分用户/设备维度**(官方 FAQ 明确拒绝),最细到 日期×应用×代码位×广告源;
-  - `revenue` = 预估收益(元,所有 ADN 都有);`api_revenue` = 收益Api(各 ADN 经 Reporting
-    回传、按实时汇率折算账号币种,更接近结算),需后台为该 ADN 配置 Reporting 才有、且不支持当天;
+  - `revenue` = 排序价/竞价实时价预估(元,非结算收入);`api_revenue` = 各 ADN 经 Reporting
+    回传、按实时汇率折算账号币种的收益,需后台为该 ADN 配置 Reporting 才有、且不支持当天;
   - 「今天」与「今天以前」必须分开查;天级跨度 ≤ 1 个月、不早于 12 个月。
 """
 from __future__ import annotations

app/models/__init__.py

@@ -13,13 +13,7 @@ from app.models.analytics_selfstat import (  # noqa: F401
 )
 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
-from app.models.cps_group import CpsGroup  # noqa: F401
-from app.models.cps_link import CpsClick, CpsLink  # noqa: F401
-from app.models.cps_order import CpsOrder  # noqa: F401
-from app.models.cps_wx_user import CpsWxUser  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
-from app.models.device import DeviceLiveness  # noqa: F401
 from app.models.coupon_state import (  # noqa: F401
     CouponClaimEvent,
     CouponClaimRecord,
@@ -27,6 +21,12 @@ from app.models.coupon_state import (  # noqa: F401
     CouponPromptEngagement,
     CouponSession,
 )
+from app.models.cps_activity import CpsActivity  # noqa: F401
+from app.models.cps_group import CpsGroup  # noqa: F401
+from app.models.cps_link import CpsClick, CpsLink  # noqa: F401
+from app.models.cps_order import CpsOrder  # noqa: F401
+from app.models.cps_wx_user import CpsWxUser  # noqa: F401
+from app.models.device import DeviceLiveness  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.guide_video import GuideVideoPlay  # noqa: F401
 from app.models.inactivity import (  # noqa: F401
@@ -39,11 +39,12 @@ from app.models.launch_confirm_sample import LaunchConfirmSample  # noqa: F401
 from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
 from app.models.notification import Notification  # noqa: F401
 from app.models.onboarding import OnboardingCompletion  # noqa: F401
-from app.models.phone_rebind_log import PhoneRebindLog  # noqa: F401
 from app.models.ops_marquee_seed import OpsMarqueeSeed  # noqa: F401
 from app.models.ops_stat_config import OpsStatConfig  # noqa: F401
+from app.models.phone_rebind_log import PhoneRebindLog  # noqa: F401
 from app.models.price_observation import PriceObservation  # noqa: F401
 from app.models.price_report import PriceReport  # noqa: F401
+from app.models.risk import BehaviorEvent, RiskIncident, SubjectRestriction  # noqa: F401
 from app.models.savings import SavingsRecord  # noqa: F401
 from app.models.signin import SigninRecord  # noqa: F401
 from app.models.store_mapping import StoreMapping  # noqa: F401

app/models/ad_pangle_revenue.py

@@ -1,15 +1,15 @@
-"""穿山甲 GroMore 天级收益报表(后台结算口径,定时拉取入库)。
+"""GroMore 天级排序价预估与 ADN Reporting 收益(定时拉取入库)。
 
 每行 = GroMore 数据 API 返回的一条「日期 × 应用 × 代码位」聚合收益(`integrations/pangle_report`
-+ `scripts/sync_pangle_revenue` 落库)。**权威/预估收益的来源**,与 `ad_ecpm_record`(客户端自报
-eCPM 折算的预估)互为对照:
++ `scripts/sync_pangle_revenue` 落库)，与 `ad_ecpm_record`(客户端 SDK eCPM 折算的预估)
+互为对照:
 
-- `revenue_yuan` ← 接口 `revenue`(预估收益,元;排序价×展示/1000,所有 ADN 都有);
-- `api_revenue_yuan` ← 接口 `api_revenue`(收益Api,元;各 ADN 经 Reporting 回传、更接近结算;
+- `revenue_yuan` ← 接口 `revenue`(排序价/竞价实时价预估,元,不是结算收入);
+- `api_revenue_yuan` ← 接口 `api_revenue`(各 ADN Reporting 回传收益,元,更接近结算;
   未配置该 ADN 的 Reporting 或查当天时为空)。
 
 ⚠️ 穿山甲不提供分用户/设备维度,故本表最细只到 日期×应用×代码位,**无法挂到逐条广告事件**;
-广告收益报表里只用于汇总/趋势级的「穿山甲后台收益」,不改逐条行的客户端预估。
+广告收益报表里只用于汇总/趋势级的 GroMore/ADN 对账,不改逐条行的客户端预估。
 """
 from __future__ import annotations
 
@@ -51,9 +51,9 @@ class AdPangleDailyRevenue(Base):
     our_code_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
     # 广告源(接口 network 数字→名,如 pangle/gdt);"" = 未分广告源的代码位汇总行(当前默认口径)。
     adn: Mapped[str] = mapped_column(String(16), nullable=False, default="")
-    # 预估收益(元)← 接口 revenue。
+    # 排序价/竞价实时价预估(元)← 接口 revenue，非结算收入。
     revenue_yuan: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
-    # 收益Api(元)← 接口 api_revenue;未配 Reporting / 当天 等情况接口不返回 → NULL。
+    # ADN Reporting API 收益(元)← api_revenue;未配 Reporting / 当天等情况不返回 → NULL。
     api_revenue_yuan: Mapped[float | None] = mapped_column(Float, nullable=True)
     # 预估 eCPM 原值(接口 ecpm,单位元/千次,**与客户端 getEcpm 的「分」不同**),参考用原样存。
     ecpm: Mapped[str | None] = mapped_column(String(32), nullable=True)

app/models/risk.py

@@ -0,0 +1,163 @@
+"""通用行为事件、风险事件与主体限制。
+
+这三张表不是“风控监控页面专用表”：
+
+- ``behavior_event`` 保存服务端权威的关键行为流水，后续可复用于安全审计、
+  漏斗核查和客诉排查；
+- ``risk_incident`` 保存规则命中后的可处置事件，负责“待处理/忽略/封禁/解除”
+  生命周期；
+- ``subject_restriction`` 保存当前生效的主体限制，统一承载设备、账号等主体的
+  业务拦截状态。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    DateTime,
+    Index,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+    true,
+)
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+_JSON = JSON().with_variant(JSONB(), "postgresql")
+
+
+class BehaviorEvent(Base):
+    """服务端权威行为流水；只增不改。"""
+
+    __tablename__ = "behavior_event"
+    __table_args__ = (
+        Index("ix_behavior_event_type_time", "event_type", "occurred_at"),
+        Index(
+            "ix_behavior_event_subject_time",
+            "subject_type",
+            "subject_id",
+            "occurred_at",
+        ),
+        Index("ix_behavior_event_user_time", "user_id", "occurred_at"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    event_type: Mapped[str] = mapped_column(String(64), nullable=False)
+    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    subject_id: Mapped[str] = mapped_column(String(128), nullable=False)
+
+    user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    device_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    device_model: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    client_ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
+    # success / failed / rejected / attempted
+    outcome: Mapped[str] = mapped_column(String(24), nullable=False, default="success")
+    reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    details: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+
+    occurred_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+
+class RiskIncident(Base):
+    """规则命中后形成的可处置风险事件。"""
+
+    __tablename__ = "risk_incident"
+    __table_args__ = (
+        UniqueConstraint(
+            "rule_code",
+            "subject_type",
+            "subject_id",
+            "window_key",
+            name="uq_risk_incident_rule_subject_window",
+        ),
+        Index("ix_risk_incident_rule_status", "rule_code", "status", "triggered_at"),
+        Index(
+            "ix_risk_incident_subject",
+            "subject_type",
+            "subject_id",
+            "triggered_at",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    rule_code: Mapped[str] = mapped_column(String(64), nullable=False)
+    event_type: Mapped[str] = mapped_column(String(64), nullable=False)
+    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    subject_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    window_key: Mapped[str] = mapped_column(String(64), nullable=False)
+
+    window_start: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    window_end: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    first_event_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    triggered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    last_event_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    event_count: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    # open / ignored / blocked / resolved
+    status: Mapped[str] = mapped_column(String(24), nullable=False, default="open")
+    action_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    handled_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    handled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    details: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+
+class SubjectRestriction(Base):
+    """主体当前限制状态；同一主体同一作用域只有一条，可封禁后再解除/重启。"""
+
+    __tablename__ = "subject_restriction"
+    __table_args__ = (
+        UniqueConstraint(
+            "subject_type",
+            "subject_id",
+            "scope",
+            name="uq_subject_restriction_subject_scope",
+        ),
+        Index(
+            "ix_subject_restriction_lookup",
+            "subject_type",
+            "subject_id",
+            "scope",
+            "active",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    subject_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    # auth_device / economic_account / all
+    scope: Mapped[str] = mapped_column(String(32), nullable=False)
+    active: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default=true()
+    )
+    reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    incident_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    created_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    revoked_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    details: Mapped[dict | None] = mapped_column(_JSON, nullable=True)

app/models/wallet.py

@@ -96,15 +96,6 @@ 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(

app/repositories/ad_ecpm.py

@@ -1,8 +1,8 @@
 """广告 eCPM 上报 CRUD(内部收益统计/对账)。
 
 客户端在广告展示后(onAdShow)读到 eCPM,经鉴权接口上报,这里落库。鉴权接口已确保
-user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响业务,
-穿山甲后台报表是结算权威兜底。
+user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响发奖业务;
+汇总收入以 ADN Reporting API 和最终结算单为准。
 """
 from __future__ import annotations
 
@@ -96,7 +96,7 @@ def create_ecpm_record(
         db.rollback()
         # 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
         # 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
-        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
+        # (收入另由 ADN Reporting API 对账),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
         # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
         # 旧逻辑在此 raise 成 500(本应静默吞掉)。
         existing = _find_by_session_global(db, ad_session_id)

app/repositories/ad_pangle_revenue.py

@@ -1,12 +1,13 @@
 """穿山甲 GroMore 天级收益 读写(`ad_pangle_daily_revenue` 表)。
 
 `scripts/sync_pangle_revenue` 拉数后调 `upsert_daily_rows` 落库(同一(日期×应用×代码位×广告源)
-幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取「穿山甲后台收益」做
+幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取 GroMore/ADN 收益做
 汇总/趋势级展示。穿山甲无用户维度,故这里不涉及 user_id。
 """
 from __future__ import annotations
 
 from collections.abc import Collection
+from datetime import datetime
 from typing import Any, TypedDict
 
 from sqlalchemy import func, select
@@ -23,6 +24,7 @@ class PangleDateAgg(TypedDict):
     revenue_yuan: float
     api_revenue_yuan: float | None
     impressions: int
+    synced_at: datetime | None
 
 
 def upsert_daily_rows(db: Session, rows: list[dict[str, Any]]) -> dict[str, int]:
@@ -87,6 +89,7 @@ def aggregate_by_date(
             func.sum(AdPangleDailyRevenue.revenue_yuan),
             func.sum(AdPangleDailyRevenue.api_revenue_yuan),
             func.sum(AdPangleDailyRevenue.impressions),
+            func.max(AdPangleDailyRevenue.synced_at),
         )
         .where(
             AdPangleDailyRevenue.report_date >= date_from,
@@ -103,11 +106,12 @@ def aggregate_by_date(
         stmt = stmt.where(AdPangleDailyRevenue.our_code_id.in_(our_code_ids))
 
     out: list[PangleDateAgg] = []
-    for report_date, rev, api_rev, imp in db.execute(stmt).all():
+    for report_date, rev, api_rev, imp, synced_at in db.execute(stmt).all():
         out.append(PangleDateAgg(
             date=report_date,
             revenue_yuan=round(float(rev or 0.0), 6),
             api_revenue_yuan=(round(float(api_rev), 6) if api_rev is not None else None),
             impressions=int(imp or 0),
+            synced_at=synced_at,
         ))
     return out

app/repositories/comparison.py

@@ -55,22 +55,13 @@ def _product_names_from_items(items: list | None) -> str | None:
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
-    _pr = payload.platform_results or {}
 
-    def _is_short(r) -> bool:
-        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
-        # platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
-        info = _pr.get(r.platform_id) if r.platform_id else None
-        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
-
-    # 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
-    # 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
-    priced = [r for r in results if r.price is not None]
-    clean = [r for r in priced if not _is_short(r)]
+    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
     best = None
-    if clean:
+    priced = [r for r in results if r.price is not None]
+    if priced:
         best = min(
-            clean,
+            priced,
             key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
         )
 
@@ -205,29 +196,14 @@ def upsert_record(
 # ============================================================
 
 
-def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None
-) -> dict:
+def _derive_from_results(results: list[dict]) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
-
-    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
-    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
-    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
-    _pr = platform_results or {}
-
-    def _is_short(r: dict) -> bool:
-        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
-        pid = r.get("platform_id")
-        info = _pr.get(pid) if pid else None
-        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
-
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
     priced = [r for r in results if r.get("price") is not None]
-    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
     best = None
-    if clean:
+    if priced:
         best = min(
-            clean,
+            priced,
             key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
         )
     src_row = next((r for r in results if r.get("is_source")), None)
@@ -413,7 +389,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
+    derived = _derive_from_results(results)
     # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
     items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(

app/repositories/risk.py

@@ -0,0 +1,569 @@
+"""通用行为事件、风险规则与主体限制仓库。"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core.config_schema import (
+    RISK_COMPARE_DAILY_THRESHOLD_KEY,
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+)
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.models.risk import BehaviorEvent, RiskIncident, SubjectRestriction
+from app.repositories import app_config
+
+CN_TZ = ZoneInfo("Asia/Shanghai")
+
+RULE_SMS_HOURLY = "sms_device_hourly_5"
+RULE_ONECLICK_DAILY = "oneclick_device_daily_20"
+RULE_COMPARE_DAILY = "compare_account_daily_100"
+
+EVENT_SMS_SEND = "sms_send"
+EVENT_SMS_LOGIN = "sms_login"
+EVENT_ONECLICK_LOGIN = "oneclick_login"
+
+SCOPE_AUTH_DEVICE = "auth_device"
+SCOPE_ECONOMIC_ACCOUNT = "economic_account"
+AUTO_RESOLVED_REASON = "规则阈值调整后不再命中"
+MANUAL_RESET_REASON = "管理员重置报警计数"
+RISK_RESET_BASELINES_KEY = "risk_monitor_reset_baselines"
+
+
+@dataclass(frozen=True)
+class RuleSpec:
+    code: str
+    event_type: str
+    subject_type: str
+    threshold_key: str
+    window: str
+    count_outcomes: tuple[str, ...]
+
+
+RULES: dict[str, RuleSpec] = {
+    RULE_SMS_HOURLY: RuleSpec(
+        code=RULE_SMS_HOURLY,
+        event_type=EVENT_SMS_SEND,
+        subject_type="device",
+        threshold_key=RISK_SMS_HOURLY_THRESHOLD_KEY,
+        window="hour",
+        count_outcomes=("success",),
+    ),
+    RULE_ONECLICK_DAILY: RuleSpec(
+        code=RULE_ONECLICK_DAILY,
+        event_type=EVENT_ONECLICK_LOGIN,
+        subject_type="device",
+        threshold_key=RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+        window="day",
+        count_outcomes=("success", "failed"),
+    ),
+}
+
+RULE_THRESHOLD_KEYS: dict[str, str] = {
+    RULE_SMS_HOURLY: RISK_SMS_HOURLY_THRESHOLD_KEY,
+    RULE_ONECLICK_DAILY: RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RULE_COMPARE_DAILY: RISK_COMPARE_DAILY_THRESHOLD_KEY,
+}
+
+
+def get_rule_threshold(db: Session, rule_code: str) -> int:
+    """读取规则当前阈值；配置表为空时回退上线前的 5/20/100 默认值。"""
+    return int(app_config.get_value(db, RULE_THRESHOLD_KEYS[rule_code]))
+
+
+def utcnow() -> datetime:
+    return datetime.now(UTC)
+
+
+def _ensure_aware(value: datetime) -> datetime:
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value
+
+
+def get_rule_reset_at(db: Session, rule_code: str) -> datetime | None:
+    """返回规则最近一次全局重置时间；异常旧值按未重置处理。"""
+    row = db.get(AppConfig, RISK_RESET_BASELINES_KEY)
+    if row is None or not isinstance(row.value, dict):
+        return None
+    raw = row.value.get(rule_code)
+    if not isinstance(raw, str):
+        return None
+    try:
+        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return _ensure_aware(parsed).astimezone(UTC)
+
+
+def reset_rule_baselines(
+    db: Session,
+    *,
+    rule_codes: tuple[str, ...],
+    reset_at: datetime,
+    admin_id: int,
+) -> None:
+    """持久化全局重置基线；不删除任何行为或比价流水。"""
+    row = db.get(AppConfig, RISK_RESET_BASELINES_KEY)
+    value = dict(row.value) if row and isinstance(row.value, dict) else {}
+    timestamp = _ensure_aware(reset_at).astimezone(UTC).isoformat()
+    value.update({rule_code: timestamp for rule_code in rule_codes})
+    if row is None:
+        db.add(
+            AppConfig(
+                key=RISK_RESET_BASELINES_KEY,
+                value=value,
+                updated_by_admin_id=admin_id,
+            )
+        )
+    else:
+        row.value = value
+        row.updated_by_admin_id = admin_id
+    db.flush()
+
+
+def _window_bounds(at: datetime, window: str) -> tuple[str, datetime, datetime]:
+    local = _ensure_aware(at).astimezone(CN_TZ)
+    if window == "hour":
+        start_local = local.replace(minute=0, second=0, microsecond=0)
+        key = start_local.strftime("%Y-%m-%dT%H")
+        end_local = start_local + timedelta(hours=1)
+    elif window == "day":
+        start_local = local.replace(hour=0, minute=0, second=0, microsecond=0)
+        key = start_local.strftime("%Y-%m-%d")
+        end_local = start_local + timedelta(days=1)
+    else:  # pragma: no cover - 规则声明错误应尽早暴露
+        raise ValueError(f"unsupported risk window: {window}")
+    return key, start_local.astimezone(UTC), end_local.astimezone(UTC)
+
+
+def _event_stats(
+    db: Session,
+    spec: RuleSpec,
+    *,
+    subject_id: str,
+    start: datetime,
+    end: datetime,
+    threshold: int,
+) -> tuple[int, datetime | None, datetime | None, datetime | None]:
+    filters = (
+        BehaviorEvent.event_type == spec.event_type,
+        BehaviorEvent.subject_type == spec.subject_type,
+        BehaviorEvent.subject_id == subject_id,
+        BehaviorEvent.outcome.in_(spec.count_outcomes),
+        BehaviorEvent.occurred_at >= start,
+        BehaviorEvent.occurred_at < end,
+    )
+    count, first_at, last_at = db.execute(
+        select(
+            func.count(BehaviorEvent.id),
+            func.min(BehaviorEvent.occurred_at),
+            func.max(BehaviorEvent.occurred_at),
+        ).where(*filters)
+    ).one()
+    triggered_at = None
+    if int(count or 0) >= threshold:
+        triggered_at = db.execute(
+            select(BehaviorEvent.occurred_at)
+            .where(*filters)
+            .order_by(BehaviorEvent.occurred_at.asc(), BehaviorEvent.id.asc())
+            .offset(threshold - 1)
+            .limit(1)
+        ).scalar_one()
+    return int(count or 0), first_at, last_at, triggered_at
+
+
+def _upsert_incident(
+    db: Session,
+    *,
+    rule_code: str,
+    event_type: str,
+    subject_type: str,
+    subject_id: str,
+    window_key: str,
+    window_start: datetime,
+    window_end: datetime,
+    first_event_at: datetime,
+    triggered_at: datetime,
+    last_event_at: datetime,
+    event_count: int,
+    details: dict | None = None,
+) -> RiskIncident:
+    incident = db.execute(
+        select(RiskIncident).where(
+            RiskIncident.rule_code == rule_code,
+            RiskIncident.subject_type == subject_type,
+            RiskIncident.subject_id == subject_id,
+            RiskIncident.window_key == window_key,
+        )
+    ).scalar_one_or_none()
+    if incident is None:
+        incident = RiskIncident(
+            rule_code=rule_code,
+            event_type=event_type,
+            subject_type=subject_type,
+            subject_id=subject_id,
+            window_key=window_key,
+            window_start=window_start,
+            window_end=window_end,
+            first_event_at=first_event_at,
+            triggered_at=triggered_at,
+            last_event_at=last_event_at,
+            event_count=event_count,
+            status="open",
+            details=details,
+        )
+        # 并发首次命中可能同时插入；保存点只回滚重复 incident，不丢行为流水。
+        try:
+            with db.begin_nested():
+                db.add(incident)
+                db.flush()
+        except IntegrityError:
+            incident = db.execute(
+                select(RiskIncident).where(
+                    RiskIncident.rule_code == rule_code,
+                    RiskIncident.subject_type == subject_type,
+                    RiskIncident.subject_id == subject_id,
+                    RiskIncident.window_key == window_key,
+                )
+            ).scalar_one()
+    else:
+        reopened = incident.status == "resolved" and incident.action_reason in (
+            AUTO_RESOLVED_REASON,
+            MANUAL_RESET_REASON,
+        )
+        if reopened:
+            incident.status = "open"
+            incident.action_reason = None
+            incident.handled_by = None
+            incident.handled_at = None
+            # 同一自然窗口内重置后会复用唯一 incident；明细窗口必须同步切到新基线，
+            # 否则展开时会把重置前的旧流水也混进来。
+            incident.window_start = window_start
+            incident.window_end = window_end
+        incident.first_event_at = first_event_at
+        if incident.status == "open":
+            incident.triggered_at = triggered_at
+        incident.last_event_at = last_event_at
+        incident.event_count = event_count
+        # 已忽略/封禁的事件只更新事实数据，不重新打开。
+        if details:
+            incident.details = {**(incident.details or {}), **details}
+    return incident
+
+
+def evaluate_behavior_rule(
+    db: Session, *, rule_code: str, subject_id: str, at: datetime
+) -> RiskIncident | None:
+    spec = RULES[rule_code]
+    threshold = get_rule_threshold(db, rule_code)
+    window_key, window_start, end = _window_bounds(at, spec.window)
+    reset_at = get_rule_reset_at(db, rule_code)
+    start = max(window_start, reset_at) if reset_at else window_start
+    count, first_at, last_at, triggered_at = _event_stats(
+        db,
+        spec,
+        subject_id=subject_id,
+        start=start,
+        end=end,
+        threshold=threshold,
+    )
+    if count < threshold or first_at is None or last_at is None or triggered_at is None:
+        return None
+    return _upsert_incident(
+        db,
+        rule_code=spec.code,
+        event_type=spec.event_type,
+        subject_type=spec.subject_type,
+        subject_id=subject_id,
+        window_key=window_key,
+        window_start=start,
+        window_end=end,
+        first_event_at=first_at,
+        triggered_at=triggered_at,
+        last_event_at=last_at,
+        event_count=count,
+    )
+
+
+def reconcile_behavior_rule(
+    db: Session,
+    *,
+    rule_code: str,
+    at: datetime | None = None,
+    commit: bool = True,
+) -> int:
+    """按当前阈值重算短信/一键登录当前窗口，并收起已不再命中的待处理告警。"""
+    spec = RULES[rule_code]
+    current = at or utcnow()
+    threshold = get_rule_threshold(db, rule_code)
+    window_key, window_start, end = _window_bounds(current, spec.window)
+    reset_at = get_rule_reset_at(db, rule_code)
+    start = max(window_start, reset_at) if reset_at else window_start
+    filters = (
+        BehaviorEvent.event_type == spec.event_type,
+        BehaviorEvent.subject_type == spec.subject_type,
+        BehaviorEvent.outcome.in_(spec.count_outcomes),
+        BehaviorEvent.occurred_at >= start,
+        BehaviorEvent.occurred_at < end,
+    )
+    qualifying_rows = db.execute(
+        select(
+            BehaviorEvent.subject_id,
+            func.max(BehaviorEvent.occurred_at),
+        )
+        .where(*filters)
+        .group_by(BehaviorEvent.subject_id)
+        .having(func.count(BehaviorEvent.id) >= threshold)
+    ).all()
+    qualifying = {str(subject_id) for subject_id, _ in qualifying_rows}
+    for subject_id, last_at in qualifying_rows:
+        evaluate_behavior_rule(
+            db,
+            rule_code=rule_code,
+            subject_id=str(subject_id),
+            at=last_at or current,
+        )
+
+    open_incidents = db.scalars(
+        select(RiskIncident).where(
+            RiskIncident.rule_code == rule_code,
+            RiskIncident.window_key == window_key,
+            RiskIncident.status == "open",
+        )
+    ).all()
+    for incident in open_incidents:
+        if incident.subject_id not in qualifying:
+            incident.status = "resolved"
+            incident.action_reason = AUTO_RESOLVED_REASON
+            incident.handled_by = None
+            incident.handled_at = utcnow()
+    if commit:
+        db.commit()
+    return len(qualifying)
+
+
+def record_behavior_event(
+    db: Session,
+    *,
+    event_type: str,
+    subject_type: str,
+    subject_id: str,
+    user_id: int | None = None,
+    device_id: str | None = None,
+    device_model: str | None = None,
+    phone: str | None = None,
+    client_ip: str | None = None,
+    outcome: str = "success",
+    reason: str | None = None,
+    details: dict | None = None,
+    occurred_at: datetime | None = None,
+    evaluate_rule: str | None = None,
+    commit: bool = True,
+) -> BehaviorEvent:
+    at = occurred_at or utcnow()
+    event = BehaviorEvent(
+        event_type=event_type,
+        subject_type=subject_type,
+        subject_id=subject_id,
+        user_id=user_id,
+        device_id=device_id,
+        device_model=device_model,
+        phone=phone,
+        client_ip=client_ip,
+        outcome=outcome,
+        reason=reason,
+        details=details,
+        occurred_at=at,
+    )
+    db.add(event)
+    db.flush()
+    if evaluate_rule:
+        evaluate_behavior_rule(db, rule_code=evaluate_rule, subject_id=subject_id, at=at)
+    if commit:
+        db.commit()
+        db.refresh(event)
+    return event
+
+
+def sync_compare_incident(
+    db: Session,
+    *,
+    user_id: int,
+    at: datetime,
+    threshold: int | None = None,
+    commit: bool = True,
+) -> RiskIncident | None:
+    effective_threshold = threshold or get_rule_threshold(db, RULE_COMPARE_DAILY)
+    # comparison_record 的既有写入口统一落“北京时间 naive”时间；这里必须沿用同一
+    # 口径，否则 SQLite/PG session timezone 不同时会把凌晨记录算到前一天。
+    local = at.astimezone(CN_TZ).replace(tzinfo=None) if at.tzinfo else at
+    window_start = local.replace(hour=0, minute=0, second=0, microsecond=0)
+    end = window_start + timedelta(days=1)
+    window_key = window_start.strftime("%Y-%m-%d")
+    reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
+    reset_local = (
+        reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
+    )
+    start = max(window_start, reset_local) if reset_local else window_start
+    filters = (
+        ComparisonRecord.user_id == user_id,
+        ComparisonRecord.created_at >= start,
+        ComparisonRecord.created_at < end,
+    )
+    count, first_at, last_at = db.execute(
+        select(
+            func.count(ComparisonRecord.id),
+            func.min(ComparisonRecord.created_at),
+            func.max(ComparisonRecord.created_at),
+        ).where(*filters)
+    ).one()
+    if int(count or 0) < effective_threshold or first_at is None or last_at is None:
+        return None
+    triggered_at = db.execute(
+        select(ComparisonRecord.created_at)
+        .where(*filters)
+        .order_by(ComparisonRecord.created_at.asc(), ComparisonRecord.id.asc())
+        .offset(effective_threshold - 1)
+        .limit(1)
+    ).scalar_one()
+    incident = _upsert_incident(
+        db,
+        rule_code=RULE_COMPARE_DAILY,
+        event_type="compare_start",
+        subject_type="user",
+        subject_id=str(user_id),
+        window_key=window_key,
+        window_start=start,
+        window_end=end,
+        first_event_at=first_at,
+        triggered_at=triggered_at,
+        last_event_at=last_at,
+        event_count=int(count),
+    )
+    if commit:
+        db.commit()
+        db.refresh(incident)
+    return incident
+
+
+def reconcile_compare_rule(
+    db: Session, *, at: datetime | None = None, commit: bool = True
+) -> int:
+    """按当前阈值重算北京时间当日比价告警，并收起不再命中的待处理告警。"""
+    current = (at or utcnow()).astimezone(CN_TZ).replace(tzinfo=None)
+    window_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
+    end = window_start + timedelta(days=1)
+    window_key = window_start.strftime("%Y-%m-%d")
+    reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
+    reset_local = (
+        reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
+    )
+    start = max(window_start, reset_local) if reset_local else window_start
+    threshold = get_rule_threshold(db, RULE_COMPARE_DAILY)
+    rows = db.execute(
+        select(
+            ComparisonRecord.user_id,
+            func.max(ComparisonRecord.created_at),
+        )
+        .where(
+            ComparisonRecord.user_id.is_not(None),
+            ComparisonRecord.created_at >= start,
+            ComparisonRecord.created_at < end,
+        )
+        .group_by(ComparisonRecord.user_id)
+        .having(func.count(ComparisonRecord.id) >= threshold)
+    ).all()
+    qualifying = {str(user_id) for user_id, _ in rows}
+    for user_id, last_at in rows:
+        sync_compare_incident(
+            db,
+            user_id=int(user_id),
+            at=last_at or current,
+            threshold=threshold,
+            commit=False,
+        )
+
+    open_incidents = db.scalars(
+        select(RiskIncident).where(
+            RiskIncident.rule_code == RULE_COMPARE_DAILY,
+            RiskIncident.window_key == window_key,
+            RiskIncident.status == "open",
+        )
+    ).all()
+    for incident in open_incidents:
+        if incident.subject_id not in qualifying:
+            incident.status = "resolved"
+            incident.action_reason = AUTO_RESOLVED_REASON
+            incident.handled_by = None
+            incident.handled_at = utcnow()
+    if commit:
+        db.commit()
+    return len(qualifying)
+
+
+def get_active_restriction(
+    db: Session, *, subject_type: str, subject_id: str, scope: str
+) -> SubjectRestriction | None:
+    return db.execute(
+        select(SubjectRestriction).where(
+            SubjectRestriction.subject_type == subject_type,
+            SubjectRestriction.subject_id == subject_id,
+            SubjectRestriction.scope.in_((scope, "all")),
+            SubjectRestriction.active.is_(True),
+        )
+    ).scalar_one_or_none()
+
+
+def is_restricted(db: Session, *, subject_type: str, subject_id: str, scope: str) -> bool:
+    return get_active_restriction(
+        db, subject_type=subject_type, subject_id=subject_id, scope=scope
+    ) is not None
+
+
+def activate_restriction(
+    db: Session,
+    *,
+    subject_type: str,
+    subject_id: str,
+    scope: str,
+    reason: str,
+    incident_id: int | None,
+    admin_id: int,
+) -> SubjectRestriction:
+    restriction = db.execute(
+        select(SubjectRestriction).where(
+            SubjectRestriction.subject_type == subject_type,
+            SubjectRestriction.subject_id == subject_id,
+            SubjectRestriction.scope == scope,
+        )
+    ).scalar_one_or_none()
+    if restriction is None:
+        restriction = SubjectRestriction(
+            subject_type=subject_type,
+            subject_id=subject_id,
+            scope=scope,
+        )
+        db.add(restriction)
+    restriction.active = True
+    restriction.reason = reason
+    restriction.incident_id = incident_id
+    restriction.created_by = admin_id
+    restriction.created_at = utcnow()
+    restriction.revoked_by = None
+    restriction.revoked_at = None
+    db.flush()
+    return restriction
+
+
+def revoke_restriction(
+    db: Session, *, restriction: SubjectRestriction, admin_id: int
+) -> None:
+    restriction.active = False
+    restriction.revoked_by = admin_id
+    restriction.revoked_at = utcnow()
+    db.flush()

app/repositories/wallet.py

@@ -35,7 +35,6 @@ from app.services import notification_events
 _WX_STATE_SUCCESS = "SUCCESS"
 _WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
 _WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM"  # 用户还没在微信确认页确认
-_WITHDRAW_ACTIVE_STATUSES = {"reviewing", "pending"}
 # 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
 # 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提
 # (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。
@@ -69,10 +68,6 @@ class InsufficientCashError(Exception):
     """现金余额不足。"""
 
 
-class WithdrawTooFrequentError(Exception):
-    """提现申请过于频繁,或已有未完成提现单。"""
-
-
 class WithdrawTierUnavailableError(Exception):
     """该档位今日不可提:次数已满,或今天已选了其他额度(7-9 福利页档位规则)。"""
 
@@ -755,17 +750,8 @@ 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
-
     # 福利页档位闸(7-9):coin_cash 只能提预设档位,且该档今日可提(服务端权威口径,防绕过
-    # 客户端刷)。放在幂等返回/在途互斥之后:同号重试仍原样返回旧单,不被档位闸误杀。
+    # 客户端刷)。放在幂等返回之后:同号重试仍原样返回旧单,不被档位闸误杀。
     # allow_sub_min(0.01 调试直发)保持原样放行,不受档位约束;invite_cash 本轮无档位概念不校验。
     if source == "coin_cash" and not allow_sub_min:
         tier_state = next(
@@ -815,6 +801,7 @@ def create_withdraw(
         db.commit()
     except IntegrityError:
         db.rollback()
+        # 唯一冲突只可能来自 out_bill_no 幂等键并发重试:原样返回既有单;否则未知冲突,上抛。
         existing = db.execute(
             select(WithdrawOrder).where(
                 WithdrawOrder.out_bill_no == out_bill_no, WithdrawOrder.user_id == user_id
@@ -822,14 +809,6 @@ def create_withdraw(
         ).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  # 待管理员审核;**不在此处打款**

app/schemas/auth.py

@@ -11,7 +11,6 @@ from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field
 
-
 # ===== 用户对外信息 =====
 
 class UserOut(BaseModel):
@@ -65,6 +64,9 @@ class JverifyLoginRequest(BaseModel):
         "", max_length=64,
         description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
     )
+    device_model: str = Field(
+        "", max_length=128, description="客户端设备型号快照,用于登录安全审计"
+    )
 
 
 # ===== 短信验证码 =====
@@ -75,6 +77,9 @@ class SmsSendRequest(BaseModel):
         "", max_length=64,
         description="硬件级设备标识(Android ANDROID_ID),用于发码防刷按 设备+IP 限流;空=按 IP 聚一桶",
     )
+    device_model: str = Field(
+        "", max_length=128, description="客户端设备型号快照,用于短信安全审计"
+    )
 
 
 class SmsSendResponse(BaseModel):
@@ -90,6 +95,9 @@ class SmsLoginRequest(BaseModel):
         "", max_length=64,
         description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
     )
+    device_model: str = Field(
+        "", max_length=128, description="客户端设备型号快照,用于短信验证安全审计"
+    )
 
 
 # ===== Refresh =====

deploy/pangle-revenue.md

@@ -1,13 +1,13 @@
 # 穿山甲 GroMore 收益拉取 定时任务 — 运维手册
 
-> 对象:维护「每天拉穿山甲后台收益入库」这套定时任务的同事。
+> 对象:维护「每天拉 GroMore / ADN 收益入库」这套定时任务的同事。
 > 🔒 服务器登录信息见**私密交接清单**,不入库。
 
 ## 它是什么
-admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查穿山甲**。穿山甲只通过 GroMore 数据 API 给数、且 **T+1**(次日约 10:00 出昨天的数),所以每天得拉一次入库,报表才会往前走。
+admin「广告收益报表」里的 GroMore / ADN 收益读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查询**。GroMore 的 T+1 初值约 10:00 可用，但第三方 ADN Reporting 数据可能到 13:50 才更新，所以需要早晚各拉一次。
 
-- 每天 10:30 跑一轮 `scripts/sync_pangle_revenue.py`,默认 `--days 3` 回补近 3 天。
-- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(预估)+ `api_revenue`(结算口径)。
+- 每天 10:30 拉初值、14:30 拉日终值，均由 `scripts/sync_pangle_revenue.py` 以 `--days 3` 回补近 3 天。
+- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(排序价预估)+ `api_revenue`(ADN Reporting 回传，更接近结算)。
 - **幂等 upsert**:同一(日期×应用×代码位)重跑只覆盖、不重复,故回补 / 重跑 / catch-up 都安全。
 - 穿山甲无用户/设备维度 → 只能落「汇总/趋势级」,报表带 user_id 过滤时这块收益置空(显示「-」)。
 
@@ -30,15 +30,15 @@ admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**
 ```bash
 sudo cp deploy/pangle-revenue.{service,timer} /etc/systemd/system/
 sudo systemctl daemon-reload && sudo systemctl enable --now pangle-revenue.timer
-systemctl list-timers pangle-revenue.timer   # 确认下次触发时间(应是次日 10:30)
+systemctl list-timers pangle-revenue.timer   # 确认下次触发时间(10:30 或 14:30)
 ```
 
 ## 怎么看健康 / 手动跑一次
 ```bash
-sudo systemctl start pangle-revenue.service     # 立即手动跑一轮(不等 10:30)
-journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 预估收益合计
+journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 收益合计
+sudo systemctl start pangle-revenue.service     # 立即手动跑一轮
 ```
-成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;预估收益合计 ¥19.42`。
+成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;排序价预估合计 ¥19.42`。
 > 看不到收益、提示 `PANGLE_REPORT_* 未配置`→ 回「上线前置」补 `.env`;报 118 → 子账号没授「查看全部数据」。
 
 ## 本机 Windows 开发(无 systemd)
@@ -57,11 +57,11 @@ journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入
 - `--start / --end`:指定闭区间(跨度 ≤ 31 天,接口上限 1 个月,超了报 114)。
 
 ## 注意事项
-- **触发时间**:`OnCalendar=*-*-* 10:30:00`。穿山甲 ~10:00 出数,故别早于 10:00 跑(会拉到空/不全)。
+- **触发时间**:10:30 提供初值，14:30 覆盖为日终值；报表只把 D+1 14:00 后同步的数据标记为日终。
 - **catch-up**:`Persistent=true` 补跑错过的那一轮;叠加 `--days 3`,漏一两天重新触发即自愈。
 - **今天 / 今天以前要分开查**:脚本默认只拉昨天及更早,不混查今天(接口约束),无需关心。
 - **join key 是 `ad_unit_id`(我们配的 104xxx)不是 `code_id`**:`code_id` 是底层各 ADN 代码位,对不上口径;`ad_unit_id='-1'` 是未归因桶。改维度时务必注意(详见脚本头注释)。
-- **`api_revenue` 很稀疏**:测试应用 ADN 没配 Reporting → 全 0,仅 prod 个别位有;`revenue`(预估)才是稳的主力。
+- **`api_revenue` 依赖 ADN Reporting 配置**:未配置的测试应用可能为空或 0；`revenue` 只是排序价估算，不能当结算收入。
 - **DB 无关**:sqlite / postgres 均可(upsert 逐行 select-then-write,不像美团 ETL 需要 PG)。
 - **别和别的触发方式双跑**:本 systemd timer 与「手动 cron / 进程内任务」二选一,虽幂等不会重复入库,纯属多余。
 - **改脚本 / 改部署**:走 git + PR,由有 root 的人部署。

deploy/pangle-revenue.service

@@ -1,5 +1,5 @@
-# 每天拉穿山甲 GroMore T+1 天级收益入库 —— 单轮跑,由 pangle-revenue.timer 每天 10:30 触发。
-# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的「穿山甲后台收益(T+1)」区块。
+# 拉 GroMore T+1 天级收益入库 —— 单轮跑,由 timer 每天 10:30、14:30 触发。
+# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的 GroMore/ADN 对账区块。
 #
 # 仅用于 Linux 服务器;本机 Windows 开发无 systemd,直接手动跑脚本即可:
 #   .venv\Scripts\python -m scripts.sync_pangle_revenue            # 拉昨天(北京时间)

deploy/pangle-revenue.timer

@@ -1,11 +1,12 @@
-# 每天 10:30 触发一次穿山甲 GroMore T+1 收益拉取入库(Linux 服务器用)。
+# 每天 10:30 首次拉取、14:30 终值复拉 GroMore T+1 收益(Linux 服务器用)。
 # 见 pangle-revenue.service 顶部注释的部署步骤。
 [Unit]
-Description=Run Pangle GroMore daily revenue sync at 10:30
+Description=Run Pangle GroMore daily revenue sync at 10:30 and 14:30
 
 [Timer]
-# 穿山甲 T+1、次日约 10:00 出数;10:30 触发留 30min 余量。要错开整点扎堆可微调到 10:35。
+# 10:30 尽早展示初值；第三方 ADN Reporting 最晚约 13:50 更新，14:30 再拉一次作为日终值。
 OnCalendar=*-*-* 10:30:00
+OnCalendar=*-*-* 14:30:00
 # 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日);叠加 --days 3 回补,漏一两天能自愈。
 Persistent=true
 AccuracySec=1min

docs/superpowers/plans/2026-07-24-withdraw-allow-concurrent.md

@@ -0,0 +1,323 @@
+# 允许在途提现时继续提交新申请 Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 取消「同一用户同一时刻仅一笔在途提现」限制,已有 `reviewing`/`pending` 单时允许继续提交新的提现申请。
+
+**Architecture:** 该限制由两道闸共同强制——应用层 `create_withdraw` 的在途单检查(`WithdrawTooFrequentError`)与数据库分区唯一索引 `ux_withdraw_order_user_active`。彻底移除两者 + 清理随之失效的死代码;既有约束(建单先扣款、`coin_cash` 每日档位次数、`out_bill_no` 幂等)天然保留,无需改动。测试库由 `Base.metadata.create_all()` 依模型建表,故删模型内索引定义即让测试反映新 schema;另配一条 Alembic 迁移让真实库(dev/prod)落地同一变更。
+
+**Tech Stack:** FastAPI · SQLAlchemy 2.0 · Alembic · pytest · ruff
+
+规格来源:[docs/superpowers/specs/2026-07-24-withdraw-allow-concurrent-design.md](../specs/2026-07-24-withdraw-allow-concurrent-design.md)
+
+---
+
+## 关键事实(实现依据)
+
+- 当前 Alembic head:`d8dd2106e438`(新迁移的 `down_revision`)。
+- 索引出处:[`alembic/versions/withdraw_safety_indexes.py`](../../../alembic/versions/withdraw_safety_indexes.py) 同时建了两个索引——本次**只删** `ux_withdraw_order_user_active`,**保留**姊妹索引 `ux_cash_transaction_withdraw_refund_ref`(退款幂等)。
+- 档位:`WithdrawTier(50, "0.5", None, 3, False)`——50 分(0.5 元)是**常规档**,每日 3 次,非新人档;测试用它来造多笔在途。
+- 测试库走 `create_all`(见 `tests/conftest.py`),**不跑 Alembic**;故迁移的正确性由本计划单独的 upgrade/downgrade 回环验证,不由 pytest 覆盖。
+- `app/models/wallet.py` 的 `Index`/`text` 仍被其它表(行 54/186/224)使用,删本表 `__table_args__` 后**无需**清理 import。
+
+---
+
+## Task 1: 允许多笔在途提现并存(TDD:模型 + 应用层 + 端点 + 迁移,单次提交)
+
+本变更是一次原子的 schema+行为改动:模型内索引、应用层检查、Alembic 迁移相互依赖,任一缺失都会让"多笔在途"在测试库或真实库其一不成立。故作为**一个任务、一次提交**完成,内部按 TDD 分步。
+
+**Files:**
+- Test: `tests/test_withdraw.py`(新增 2 个用例)
+- Modify: `app/repositories/wallet.py`(删在途单检查 + IntegrityError 兜底瘦身 + 删死常量/异常)
+- Modify: `app/api/v1/wallet.py:225-229`(删失效的 409 处理)
+- Modify: `app/models/wallet.py:99-107`(删分区唯一索引)
+- Create: `alembic/versions/drop_withdraw_active_unique_index.py`(真实库删索引)
+
+---
+
+- [ ] **Step 1: 写两个失败测试**
+
+在 `tests/test_withdraw.py` 末尾追加(复用文件内既有 helper `_login`/`_auth`/`_seed_cash`/`_patch_userinfo`):
+
+```python
+def test_withdraw_multiple_in_flight_allowed(client, monkeypatch) -> None:
+    """取消「同时仅一单」:已有在途(reviewing)时,不同 out_bill_no 可继续提交,两单并存。"""
+    _patch_userinfo(monkeypatch, "openid_multi_inflight")
+    token = _login(client, "13800002020")
+    _seed_cash(client, token, "13800002020", 100)  # 够两笔 0.5
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r1 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billmulti00000001"},
+        headers=_auth(token),
+    )
+    assert r1.status_code == 200, r1.text
+    assert r1.json()["status"] == "reviewing"
+
+    r2 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billmulti00000002"},
+        headers=_auth(token),
+    )
+    assert r2.status_code == 200, r2.text
+    assert r2.json()["status"] == "reviewing"
+
+    # 两张在途单并存
+    r = client.get("/api/v1/wallet/withdraw-orders", headers=_auth(token))
+    reviewing = [o for o in r.json()["items"] if o["status"] == "reviewing"]
+    assert len(reviewing) == 2, r.text
+    # 余额扣两次:100-50-50=0
+    r = client.get("/api/v1/wallet/account", headers=_auth(token))
+    assert r.json()["cash_balance_cents"] == 0
+
+
+def test_withdraw_second_blocked_only_by_insufficient_cash(client, monkeypatch) -> None:
+    """并行放开后第二笔仅受余额约束:余额不足返 409「现金余额不足」,而非旧的「已有提现」拦截。"""
+    _patch_userinfo(monkeypatch, "openid_multi_insuff")
+    token = _login(client, "13800002021")
+    _seed_cash(client, token, "13800002021", 50)  # 仅够一笔 0.5
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r1 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billinsuff0000001"},
+        headers=_auth(token),
+    )
+    assert r1.status_code == 200, r1.text
+
+    r2 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billinsuff0000002"},
+        headers=_auth(token),
+    )
+    assert r2.status_code == 409, r2.text
+    assert "现金余额不足" in r2.json()["detail"]
+```
+
+- [ ] **Step 2: 运行新测试,确认失败**
+
+Run: `pytest tests/test_withdraw.py::test_withdraw_multiple_in_flight_allowed tests/test_withdraw.py::test_withdraw_second_blocked_only_by_insufficient_cash -q`
+Expected: 两条 FAIL —— `multiple_in_flight` 因第二笔被拦返回 409(期望 200);`second_blocked` 因返回的 409 detail 是「已有提现申请正在审核或打款中」而非「现金余额不足」。
+
+- [ ] **Step 3: 应用层删在途单互斥检查**(`app/repositories/wallet.py` · `create_withdraw`)
+
+删掉在途单预检查(它紧邻档位闸注释之前):
+
+old:
+```python
+    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
+
+    # 福利页档位闸(7-9):coin_cash 只能提预设档位,且该档今日可提(服务端权威口径,防绕过
+```
+new:
+```python
+    # 福利页档位闸(7-9):coin_cash 只能提预设档位,且该档今日可提(服务端权威口径,防绕过
+```
+
+- [ ] **Step 4: 应用层给 IntegrityError 兜底瘦身**(同函数末尾 commit 处)
+
+索引移除后不会再因在途单触发唯一冲突,只保留 `out_bill_no` 幂等重试分支:
+
+old:
+```python
+    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
+```
+new:
+```python
+    except IntegrityError:
+        db.rollback()
+        # 唯一冲突只可能来自 out_bill_no 幂等键并发重试:原样返回既有单;否则未知冲突,上抛。
+        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
+        raise
+```
+
+- [ ] **Step 5: 删死常量 `_WITHDRAW_ACTIVE_STATUSES`**(`app/repositories/wallet.py` 模块顶部)
+
+只删常量行,保留其后属于 `_NEWBIE_TIER_HELD_STATUSES` 的注释:
+
+old:
+```python
+_WITHDRAW_ACTIVE_STATUSES = {"reviewing", "pending"}
+# 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
+```
+new:
+```python
+# 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
+```
+
+- [ ] **Step 6: 删死异常类 `WithdrawTooFrequentError`**(`app/repositories/wallet.py`)
+
+old:
+```python
+class WithdrawTooFrequentError(Exception):
+    """提现申请过于频繁,或已有未完成提现单。"""
+
+
+class WithdrawTierUnavailableError(Exception):
+```
+new:
+```python
+class WithdrawTierUnavailableError(Exception):
+```
+
+- [ ] **Step 7: 删端点内失效的 409 处理**(`app/api/v1/wallet.py` · `withdraw`)
+
+old:
+```python
+    except crud_wallet.WechatNotBoundError as 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.WithdrawTierUnavailableError as e:
+```
+new:
+```python
+    except crud_wallet.WechatNotBoundError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先绑定微信") from e
+    except crud_wallet.WithdrawTierUnavailableError as e:
+```
+
+- [ ] **Step 8: 删模型内分区唯一索引**(`app/models/wallet.py` · `WithdrawOrder`)
+
+old:
+```python
+    __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')"),
+        ),
+    )
+```
+new:
+```python
+    __tablename__ = "withdraw_order"
+```
+
+- [ ] **Step 9: 运行新测试,确认通过**
+
+Run: `pytest tests/test_withdraw.py::test_withdraw_multiple_in_flight_allowed tests/test_withdraw.py::test_withdraw_second_blocked_only_by_insufficient_cash -q`
+Expected: 2 passed。
+
+- [ ] **Step 10: 建 Alembic 迁移(真实库删索引)**
+
+创建 `alembic/versions/drop_withdraw_active_unique_index.py`:
+
+```python
+"""drop withdraw active-order partial unique index (allow multiple in-flight withdrawals)
+
+Revision ID: drop_withdraw_active_unique_index
+Revises: d8dd2106e438
+Create Date: 2026-07-24 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "drop_withdraw_active_unique_index"
+down_revision: Union[str, Sequence[str], None] = "d8dd2106e438"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 取消「同一用户同一时刻仅一笔在途提现」:允许 reviewing/pending 并存。
+    # 仅删本索引;姊妹索引 ux_cash_transaction_withdraw_refund_ref(退款幂等)保持不动。
+    op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order")
+
+
+def downgrade() -> None:
+    # 回滚重建分区唯一索引。注意:若届时某用户已有 ≥2 张在途单,重建会因唯一冲突失败——
+    # 属预期的回滚代价(取消限制后本就允许多单),需先人工收敛在途单再回滚。
+    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')"),
+    )
+```
+
+- [ ] **Step 11: 验证迁移 upgrade + 回环 downgrade/upgrade**
+
+Run: `alembic upgrade head`
+Expected: 输出应用 `drop_withdraw_active_unique_index`,无报错。
+
+Run: `alembic downgrade -1 && alembic upgrade head`
+Expected: downgrade 重建索引、upgrade 再次删除,均成功(dev 库每用户在途单 ≤1,不会触发唯一冲突)。
+
+- [ ] **Step 12: 全量测试 + Lint**
+
+Run: `pytest -q`
+Expected: 全绿(既有 `test_withdraw_idempotent_same_bill_no`、`tests/test_withdraw_tiers.py`、`tests/test_invite_cash_withdraw.py` 均不受影响)。
+
+Run: `ruff check .`
+Expected: 无新增告警(死常量/异常已连同引用一并删除)。
+
+- [ ] **Step 13: 提交**
+
+```bash
+git add tests/test_withdraw.py app/repositories/wallet.py app/api/v1/wallet.py app/models/wallet.py alembic/versions/drop_withdraw_active_unique_index.py
+git commit -m "feat(withdraw): 允许在途提现时继续提交新申请
+
+取消「同一用户同时仅一笔在途提现」限制:删应用层在途单检查 +
+删 DB 分区唯一索引 ux_withdraw_order_user_active + 清理死代码
+(_WITHDRAW_ACTIVE_STATUSES / WithdrawTooFrequentError)。
+先扣款、coin_cash 每日档位次数、out_bill_no 幂等等既有约束不变。
+
+Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
+```
+
+---
+
+## 自审(spec 覆盖核对)
+
+- **R1 已有 reviewing/pending 可再提** → Step 3/4(应用层)+ Step 8/10(模型 + 迁移);`test_withdraw_multiple_in_flight_allowed` 证明。✓
+- **R2 不加硬上限** → 无新增限制;`coin_cash` 天然封顶由既有档位次数(`tests/test_withdraw_tiers.py`)保障,不改。✓
+- **R3 幂等/限额/退款/对账不回退** → `test_withdraw_idempotent_same_bill_no` 与档位/邀请现金测试留绿(Step 12);解绑退款、对账按单号维度,未触碰。✓
+- **spec §4 四处改动** → Step 3/4/5/6(repo)、Step 7(api)、Step 8(model)、Step 10(迁移)一一对应。✓
+- **spec §6 客户端注意点** → 属跨仓 App 事项,文档已记,本计划无对应代码任务(有意为之)。✓
+- **占位符扫描**:迁移 `revision` / `down_revision`(`d8dd2106e438`)均为具体值,无 TBD。✓
+- **命名一致性**:`WithdrawTooFrequentError`/`_WITHDRAW_ACTIVE_STATUSES` 的删除点与引用点全覆盖(全仓仅这 4 处);`ux_withdraw_order_user_active` 在模型/迁移中拼写一致。✓

docs/superpowers/specs/2026-07-24-withdraw-allow-concurrent-design.md

@@ -0,0 +1,119 @@
+# 允许在途提现时继续提交新的提现申请 设计
+
+- **日期**:2026-07-24
+- **状态**:Draft — 待评审
+- **所属**:app-server(`app/`),含一处 Alembic 迁移;另有一条 Android/客户端注意点(非本次后端工作)
+- **一句话**:取消「同一用户同一时刻只能有一笔在途提现」的限制 —— 已有 `reviewing`(待审核)或 `pending`(打款在途)提现单时,允许再次发起新的提现申请。
+
+---
+
+## 1. 背景与目标
+
+现状:用户发起提现后,在管理员审核通过并打款完成之前(`reviewing` / `pending`),**无法再发起第二笔提现**,会收到 409「已有提现申请正在审核或打款中,请处理完成后再申请」。人工审核有延迟时,用户被卡住、体验差。
+
+**目标**:放开该限制,已有在途提现单时仍可继续提交新的提现申请。
+
+### 非目标(本期不做)
+
+- 不改「先扣款」模型(提现建单即原子扣现金,天然防超提)。
+- 不改 `coin_cash` 的每日档位次数限制(仍是独立的限额闸)。
+- 不加任何新的并发/次数硬上限(见 §3 决策)。
+- 不改 admin 审核/打款/对账逻辑(其全部按单号维度操作,天然支持一人多单)。
+
+---
+
+## 2. 需求
+
+| # | 需求 | 落地 |
+|---|---|---|
+| R1 | 已有 `reviewing` 或 `pending` 单时可再次发起提现 | 删除应用层在途单互斥检查 + 删除 DB 分区唯一索引(§4) |
+| R2 | 不引入新的并发上限 | 仅靠既有约束:先扣款余额 + `coin_cash` 每日档位次数(§5) |
+| R3 | 既有幂等/限额/退款/对账行为不回退 | 保留 `out_bill_no` 幂等、档位闸、解绑退款、对账(§5) |
+
+---
+
+## 3. 决策记录(来自评审问答)
+
+| 决策点 | 结论 | 理由 |
+|---|---|---|
+| **放行范围** | `reviewing` 与 `pending` **两种在途状态都放行**,彻底取消「同时仅一单」 | 需求即"存在在途提现时可继续提交";人工审核延迟不应卡住用户 |
+| **并发上限** | **不加硬上限** | 现金**先扣款**→每笔各需自己的余额,不会超提;`coin_cash` 每日档位次数已是天然上限;`invite_cash` 仅受余额约束,可接受 |
+| **实现方式** | **彻底移除**(删检查 + 删索引 + 清理死代码),非配置开关 | 决策明确且无回滚诉求,YAGNI;避免留半死代码与多余配置项 |
+| **已失效异常/文案** | 删除 `WithdrawTooFrequentError` 及端点的 409 处理 | 全仓仅这 4 处引用(§4),移除后无残留 |
+
+### 已知取舍(可接受)
+
+- **多笔 `pending` × 未开免确认**:每笔 `pending` 会各自返回一个微信确认页 `package_info`。用户若**未开启免确认**,可能同时存在多笔待确认。属客户端交互问题(§6),后端行为正确;开启免确认后直接到账、无此问题。
+- **回滚风险**:迁移 `downgrade` 会重建分区唯一索引;若届时某用户已有 ≥2 张在途单,重建会失败 —— 属预期的回滚代价,在迁移里注释说明。
+
+---
+
+## 4. 现状机制与改动点
+
+「同一时刻仅一笔在途提现」由**两道闸**共同强制,均以 `status IN ('reviewing','pending')` 为口径:
+
+1. **应用层** — [`app/repositories/wallet.py:758-765`](../../../app/repositories/wallet.py) `create_withdraw` 内:查在途单 → `raise WithdrawTooFrequentError` → 端点转 409。
+2. **数据库层** — [`app/models/wallet.py:99-107`](../../../app/models/wallet.py) 的**分区唯一索引** `ux_withdraw_order_user_active`(`user_id WHERE status IN ('reviewing','pending')`)。这是硬约束,`create_withdraw` 的 `IntegrityError` 兜底([`:814-833`](../../../app/repositories/wallet.py))即捕获它。
+
+> 提现单状态机:`reviewing`(待审核,建单即扣现金)→`pending`(审核通过、打款在途)→`success`/`failed`;或 `reviewing`→`rejected`(已退款)。
+
+### 改动清单(4 处代码 + 1 个迁移)
+
+**① `app/repositories/wallet.py` · `create_withdraw`**
+- 删除在途单互斥检查([`:758-765`](../../../app/repositories/wallet.py)):`select WithdrawOrder.id WHERE status IN (_WITHDRAW_ACTIVE_STATUSES)` → `raise WithdrawTooFrequentError`。
+- `IntegrityError` 兜底([`:814-833`](../../../app/repositories/wallet.py)):**保留** `out_bill_no` 幂等重试分支([`:816-824`](../../../app/repositories/wallet.py));**删除**其中的在途单二次检查分支([`:825-832`](../../../app/repositories/wallet.py))(索引移除后不会再因在途单触发 `IntegrityError`);末尾其余情况原样 `raise`(仅剩 `out_bill_no` 唯一冲突等,正常不该出现)。
+- 删除模块常量 `_WITHDRAW_ACTIVE_STATUSES`([`:38`](../../../app/repositories/wallet.py))与异常类 `WithdrawTooFrequentError`([`:72-73`](../../../app/repositories/wallet.py))。`_NEWBIE_TIER_HELD_STATUSES`(档位资格判定,含 `success`)与本改动无关,**保留**。
+
+**② `app/models/wallet.py` · `WithdrawOrder`**
+- 删除 `__table_args__` 中的分区唯一索引 `ux_withdraw_order_user_active`([`:99-107`](../../../app/models/wallet.py))。该表 `__table_args__` 仅此一项,整块移除。
+
+**③ `app/api/v1/wallet.py` · `withdraw` 端点**
+- 删除 `except crud_wallet.WithdrawTooFrequentError`([`:225-229`](../../../app/api/v1/wallet.py))这段已失效的 409 处理。
+
+**④ 新增 Alembic 迁移** `alembic/versions/<...>_drop_withdraw_active_unique_index.py`
+- `down_revision` = 当前 head(实现时确定)。
+- `upgrade`:`op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order")`。
+- `downgrade`:`op.create_index("ux_withdraw_order_user_active", "withdraw_order", ["user_id"], unique=True, sqlite_where=text("status IN ('reviewing','pending')"), postgresql_where=text("status IN ('reviewing','pending')"))`,并注释"若已有用户存在多张在途单则重建失败,属预期回滚代价"。
+- 索引的 drop/create 为具名操作,SQLite/PG 均无需 `render_as_batch` 重建表。
+
+---
+
+## 5. 天然保持不变(无需改动)的约束
+
+| 约束 | 为何仍成立 |
+|---|---|
+| **不会超提** | 建单即原子扣现金(`_try_deduct_cash`,余额不足影响 0 行 → `InsufficientCashError`);每笔并行单各需自己的余额 |
+| **`coin_cash` 每日档位次数** | `withdraw_tier_states` 的档位闸独立于在途单检查:常规档按**当天发起即计入(任意状态,含被拒/失败,按 `created_at` 北京日)**、每档每日限次(0.5×3 / 10×1 / 20×1)且当天只选一档;新人档(0.1/0.3)按 `reviewing/pending/success` 一次性占用。故即便并行,`coin_cash` 单日在途仍被档位天然封顶(至多 3 笔 0.5) |
+| **`out_bill_no` 幂等** | 幂等分支在被删检查之前,同号重试仍原样返回旧单、不重复扣款 |
+| **解绑退款** | `refund_reviewing_withdraws_on_unbind` 已 `for` 遍历该用户**所有** `reviewing` 单,天然支持多单 |
+| **admin 审核 / 打款 / 对账** | `approve_withdraw`/`reject_withdraw`/`refresh_withdraw_status`/对账全按 `out_bill_no` 单号维度,不假设一人一单 |
+
+---
+
+## 6. 客户端/App 团队注意点(跨仓,非本次后端工作)
+
+> - 允许多笔并行后,每次提交都要生成**新的 `out_bill_no`**(复用旧号会命中幂等、返回旧单)。
+> - 用户**未开启免确认**时,多笔 `pending` 会各自返回一个微信确认页 `package_info`,App 需能处理/串行多笔待确认。开启免确认后直接到账、无此问题。
+
+---
+
+## 7. 测试计划(`tests/test_withdraw.py`,沿用现有 helper)
+
+- **新增 · 多笔在途放行**:同一用户 seed ≥1 元现金,用**两个不同 `out_bill_no`** 各提 0.5 元(在 0.5 元档 3 次/天限额内)→ 两次均 200 且 `status=reviewing`;`/withdraw-orders` 返回 2 条;余额正确扣两次(1 元 → 0)。
+- **新增 · 第二笔余额不足**:seed 0.5 元,连提两笔 0.5 元 → 第一笔 200、第二笔 409(`InsufficientCashError`),验证每笔各需自己的余额。
+- **回归 · 幂等**:`test_withdraw_idempotent_same_bill_no`(同 `out_bill_no` → 同一单、只扣一次)仍绿。
+- **回归 · 档位限额**:`tests/test_withdraw_tiers.py`(0.5 元日 3 次、第 4 次 409;跨档互斥)不受影响仍绿。
+- 沿用 `tests/conftest.py`(临时 SQLite、`RATE_LIMIT_ENABLED=false`);wxpay 网络调用全部 monkeypatch。
+
+---
+
+## 附:涉及文件清单
+
+**改动**
+- `app/repositories/wallet.py` — 删在途单检查 + `IntegrityError` 兜底瘦身 + 删 `_WITHDRAW_ACTIVE_STATUSES` / `WithdrawTooFrequentError`
+- `app/models/wallet.py` — 删分区唯一索引 `ux_withdraw_order_user_active`
+- `app/api/v1/wallet.py` — 删 `WithdrawTooFrequentError` 的 409 处理
+- `tests/test_withdraw.py` — 新增多笔在途放行 / 第二笔余额不足用例
+
+**新增**
+- `alembic/versions/<...>_drop_withdraw_active_unique_index.py` — drop 分区唯一索引(downgrade 重建)

scripts/seed_admin_revenue_mock.py

@@ -0,0 +1,211 @@
+"""生成本地 admin「广告收益」与数据大盘用的可重复 mock 数据。
+
+只处理 ``local-admin-revenue-mock-`` 前缀的数据，重跑会替换自身数据，不会触碰真实本地记录。
+会覆盖 Draw（含一条历史 feed）、福利激励视频、提现视频，以及对应的金币流水。
+
+用法：
+    python -m scripts.seed_admin_revenue_mock
+"""
+from __future__ import annotations
+
+from datetime import datetime, time, timedelta
+
+from sqlalchemy import delete, select
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ, cn_today
+from app.db.session import SessionLocal
+from app.models.ad_ecpm import AdEcpmRecord
+from app.models.ad_feed_reward import AdFeedRewardRecord
+from app.models.ad_reward import AdRewardRecord
+from app.models.user import User
+from app.models.wallet import CoinAccount, CoinTransaction
+
+PREFIX = "local-admin-revenue-mock-"
+PHONE = "19900009002"
+USERNAME = "80000009002"
+
+
+def _at(day_offset: int, hour: int, minute: int) -> datetime:
+    day = cn_today() - timedelta(days=day_offset)
+    return datetime.combine(day, time(hour, minute), tzinfo=CN_TZ)
+
+
+def _add_coin(
+    db,
+    *,
+    user_id: int,
+    amount: int,
+    biz_type: str,
+    ref_id: str,
+    created_at: datetime,
+    balance_after: int,
+) -> None:
+    db.add(CoinTransaction(
+        user_id=user_id,
+        amount=amount,
+        balance_after=balance_after,
+        biz_type=biz_type,
+        ref_id=ref_id,
+        remark="本地运营后台广告收益 Mock",
+        created_at=created_at,
+    ))
+
+
+def seed() -> dict[str, int]:
+    if settings.APP_ENV == "prod":
+        raise RuntimeError("Refusing to seed admin revenue mock data in production")
+
+    with SessionLocal() as db:
+        # 清理顺序按外键依赖从流水/奖励到展示；只碰本脚本自己的稳定前缀。
+        db.execute(delete(CoinTransaction).where(CoinTransaction.ref_id.like(f"{PREFIX}%")))
+        db.execute(delete(AdFeedRewardRecord).where(
+            AdFeedRewardRecord.client_event_id.like(f"{PREFIX}%")
+        ))
+        db.execute(delete(AdRewardRecord).where(AdRewardRecord.trans_id.like(f"{PREFIX}%")))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.like(f"{PREFIX}%")))
+
+        user = db.execute(select(User).where(User.phone == PHONE)).scalar_one_or_none()
+        if user is None:
+            user = User(
+                phone=PHONE,
+                username=USERNAME,
+                nickname="运营收益 Mock 用户",
+                register_channel="sms",
+                status="active",
+            )
+            db.add(user)
+            db.flush()
+        else:
+            user.nickname = "运营收益 Mock 用户"
+            user.status = "active"
+
+        balance = 0
+        event_count = 0
+        reward_count = 0
+        # 近四天的数据既能覆盖单日，也能覆盖近 7 天趋势与分类合计。
+        for day_offset in range(4):
+            suffix = f"d{day_offset}"
+            compare_trace = "local-invite-mock-compare-success"
+            coupon_trace = "mock-coupon-repeat-prod-second"
+
+            draw_events = [
+                ("draw", "comparison", compare_trace, 2800 + day_offset * 100, 10, 10),
+                ("draw", "coupon", coupon_trace, 1750 + day_offset * 100, 10, 28),
+            ]
+            # 历史 feed 必须被 Draw 分类一起计算，用于走查兼容逻辑。
+            if day_offset == 1:
+                draw_events.append(("feed", "coupon", coupon_trace, 1250, 11, 12))
+
+            for index, (ad_type, scene, trace_id, ecpm, hour, minute) in enumerate(draw_events, start=1):
+                session = f"{PREFIX}{suffix}-draw-{index}"
+                created_at = _at(day_offset, hour, minute)
+                db.add(AdEcpmRecord(
+                    user_id=user.id,
+                    ad_type=ad_type,
+                    feed_scene=scene,
+                    trace_id=trace_id,
+                    ad_session_id=session,
+                    adn="pangle" if index == 1 else "gdt",
+                    slot_id="mock-draw-rit",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw=str(ecpm),
+                    report_date=created_at.date().isoformat(),
+                    created_at=created_at,
+                ))
+                coin = 18 + day_offset * 2
+                db.add(AdFeedRewardRecord(
+                    client_event_id=f"{PREFIX}{suffix}-feed-reward-{index}",
+                    ad_session_id=session,
+                    user_id=user.id,
+                    reward_date=created_at.date().isoformat(),
+                    duration_seconds=20,
+                    unit_count=2,
+                    ecpm_raw=str(ecpm),
+                    adn="pangle" if index == 1 else "gdt",
+                    slot_id="mock-draw-rit",
+                    ad_type=ad_type,
+                    feed_scene=scene,
+                    trace_id=trace_id,
+                    app_env="prod",
+                    our_code_id="104098712",
+                    coin=coin,
+                    status="granted",
+                    created_at=created_at + timedelta(seconds=20),
+                ))
+                balance += coin
+                _add_coin(
+                    db,
+                    user_id=user.id,
+                    amount=coin,
+                    biz_type="feed_ad_reward",
+                    ref_id=f"{PREFIX}{suffix}-feed-coin-{index}",
+                    created_at=created_at + timedelta(seconds=20),
+                    balance_after=balance,
+                )
+                event_count += 1
+                reward_count += 1
+
+            for ad_type, ecpm, hour, coin in (
+                ("reward_video", 13200 + day_offset * 500, 14, 66),
+                ("withdrawal_video", 32000 + day_offset * 800, 18, 0),
+            ):
+                session = f"{PREFIX}{suffix}-{ad_type}"
+                created_at = _at(day_offset, hour, 6)
+                db.add(AdEcpmRecord(
+                    user_id=user.id,
+                    ad_type=ad_type,
+                    ad_session_id=session,
+                    adn="ks" if ad_type == "reward_video" else "baidu",
+                    slot_id="mock-video-rit",
+                    app_env="prod",
+                    our_code_id="104099389",
+                    ecpm_raw=str(ecpm),
+                    report_date=created_at.date().isoformat(),
+                    created_at=created_at,
+                ))
+                event_count += 1
+                if ad_type == "reward_video":
+                    db.add(AdRewardRecord(
+                        trans_id=f"{PREFIX}{suffix}-reward-video",
+                        user_id=user.id,
+                        coin=coin,
+                        status="granted",
+                        reward_scene="reward_video",
+                        ad_session_id=session,
+                        ecpm_raw=str(ecpm),
+                        app_env="prod",
+                        our_code_id="104099389",
+                        reward_date=created_at.date().isoformat(),
+                        reward_name="Mock 福利视频",
+                        created_at=created_at + timedelta(seconds=35),
+                    ))
+                    balance += coin
+                    _add_coin(
+                        db,
+                        user_id=user.id,
+                        amount=coin,
+                        biz_type="reward_video",
+                        ref_id=f"{PREFIX}{suffix}-reward-video-coin",
+                        created_at=created_at + timedelta(seconds=35),
+                        balance_after=balance,
+                    )
+                    reward_count += 1
+
+        account = db.get(CoinAccount, user.id)
+        if account is None:
+            account = CoinAccount(user_id=user.id)
+            db.add(account)
+        account.coin_balance = balance
+        account.total_coin_earned = balance
+        db.commit()
+        return {"events": event_count, "rewards": reward_count, "coin": balance}
+
+
+if __name__ == "__main__":
+    result = seed()
+    print(
+        "Seeded local admin revenue mock: "
+        f"{result['events']} impressions, {result['rewards']} rewards, {result['coin']} coins"
+    )

scripts/seed_risk_monitor_demo.py

@@ -0,0 +1,274 @@
+"""为本地“风控监控”页面灌入一组可重复的完整演示数据。
+
+仅允许在 APP_ENV=dev 运行。脚本会重建 ``risk-demo-*`` 前缀的数据，并创建本地
+后台账号 ``risk_demo / RiskDemo123!``，方便端到端和视觉验收。
+"""
+from __future__ import annotations
+
+from datetime import timedelta
+
+from sqlalchemy import delete, select
+
+from app.admin.repositories import admin_user as admin_repo
+from app.core.config import settings
+from app.core.security import hash_password
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
+from app.models.risk import BehaviorEvent, RiskIncident, SubjectRestriction
+from app.repositories import risk as risk_repo
+from app.repositories import user as user_repo
+
+DEMO_PREFIX = "risk-demo-"
+ADMIN_USERNAME = "risk_demo"
+ADMIN_PASSWORD = "RiskDemo123!"
+
+
+def _reset_demo(db) -> None:
+    demo_user_ids = list(
+        db.scalars(
+            select(ComparisonRecord.user_id)
+            .where(
+                ComparisonRecord.trace_id.like(f"{DEMO_PREFIX}%"),
+                ComparisonRecord.user_id.is_not(None),
+            )
+            .distinct()
+        ).all()
+    )
+    demo_incident_ids = list(
+        db.scalars(
+            select(RiskIncident.id).where(
+                (RiskIncident.subject_id.like(f"{DEMO_PREFIX}%"))
+                | (
+                    (RiskIncident.rule_code == risk_repo.RULE_COMPARE_DAILY)
+                    & (RiskIncident.subject_id.in_([str(value) for value in demo_user_ids]))
+                )
+            )
+        ).all()
+    )
+    if demo_incident_ids:
+        db.execute(
+            delete(SubjectRestriction).where(
+                SubjectRestriction.incident_id.in_(demo_incident_ids)
+            )
+        )
+    db.execute(
+        delete(SubjectRestriction).where(
+            SubjectRestriction.subject_id.like(f"{DEMO_PREFIX}%")
+        )
+    )
+    db.execute(
+        delete(RiskIncident).where(
+            (RiskIncident.subject_id.like(f"{DEMO_PREFIX}%"))
+            | (
+                (RiskIncident.rule_code == risk_repo.RULE_COMPARE_DAILY)
+                & (RiskIncident.subject_id.in_([str(value) for value in demo_user_ids]))
+            )
+        )
+    )
+    db.execute(
+        delete(BehaviorEvent).where(
+            BehaviorEvent.subject_id.like(f"{DEMO_PREFIX}%")
+        )
+    )
+    db.execute(
+        delete(ComparisonRecord).where(
+            ComparisonRecord.trace_id.like(f"{DEMO_PREFIX}%")
+        )
+    )
+    db.commit()
+
+
+def _ensure_admin(db) -> None:
+    admin = admin_repo.get_by_username(db, ADMIN_USERNAME)
+    if admin is None:
+        admin_repo.create_admin(
+            db,
+            username=ADMIN_USERNAME,
+            password=ADMIN_PASSWORD,
+            role="super_admin",
+        )
+        return
+    admin.password_hash = hash_password(ADMIN_PASSWORD)
+    admin.role = "super_admin"
+    admin.status = "active"
+    db.commit()
+
+
+def _seed_sms(db, now) -> None:
+    alerted = [
+        ("risk-demo-sms-oppo-a5", "OPPO A5", "13812343001", 9),
+        ("risk-demo-sms-xiaomi14", "Xiaomi 14", "13812343002", 7),
+        ("risk-demo-sms-iphone15", "iPhone 15", "13812343003", 5),
+    ]
+    events: list[BehaviorEvent] = []
+    for device_id, model, phone, count in alerted:
+        for index in range(count):
+            events.append(
+                BehaviorEvent(
+                    event_type=risk_repo.EVENT_SMS_SEND,
+                    subject_type="device",
+                    subject_id=device_id,
+                    device_id=device_id,
+                    device_model=model,
+                    phone=phone if index == 0 else f"138{index:08d}"[-11:],
+                    client_ip="127.0.0.1",
+                    outcome="success",
+                    details={"demo": True},
+                    occurred_at=now + timedelta(seconds=index),
+                )
+            )
+    # 其余成功下发分散到未达阈值的设备，每台最多 4 条，确保报警设备仍严格为 3 台。
+    remaining = 1284 - sum(row[3] for row in alerted)
+    for index in range(remaining):
+        device_no = index // 4
+        device_id = f"risk-demo-sms-normal-{device_no:04d}"
+        events.append(
+            BehaviorEvent(
+                event_type=risk_repo.EVENT_SMS_SEND,
+                subject_type="device",
+                subject_id=device_id,
+                device_id=device_id,
+                device_model="演示普通设备",
+                phone=f"137{index % 100_000_000:08d}",
+                client_ip="127.0.0.1",
+                outcome="success",
+                details={"demo": True},
+                occurred_at=now + timedelta(seconds=index % 1800),
+            )
+        )
+    db.add_all(events)
+    db.flush()
+    for device_id, _, phone, _ in alerted:
+        risk_repo.evaluate_behavior_rule(
+            db,
+            rule_code=risk_repo.RULE_SMS_HOURLY,
+            subject_id=device_id,
+            at=now,
+        )
+        user = user_repo.upsert_user_for_login(
+            db, phone=phone, register_channel="sms"
+        )
+        db.add(
+            BehaviorEvent(
+                event_type=risk_repo.EVENT_SMS_LOGIN,
+                subject_type="device",
+                subject_id=device_id,
+                user_id=user.id,
+                device_id=device_id,
+                phone=phone,
+                outcome="success",
+                details={"demo": True},
+                occurred_at=now + timedelta(minutes=2),
+            )
+        )
+    db.commit()
+
+
+def _seed_oneclick(db, now) -> None:
+    devices = [
+        ("risk-demo-oneclick-oppo-reno", "OPPO Reno", "13912343001", 482),
+        ("risk-demo-oneclick-vivo-y36", "vivo Y36", "13912343002", 481),
+    ]
+    for device_id, model, phone, count in devices:
+        user = user_repo.upsert_user_for_login(
+            db, phone=phone, register_channel="jverify"
+        )
+        db.add_all(
+            [
+                BehaviorEvent(
+                    event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+                    subject_type="device",
+                    subject_id=device_id,
+                    user_id=user.id,
+                    device_id=device_id,
+                    device_model=model,
+                    phone=phone,
+                    client_ip="127.0.0.1",
+                    outcome="failed" if index % 11 == 0 else "success",
+                    reason="运营商校验失败" if index % 11 == 0 else None,
+                    details={"demo": True},
+                    occurred_at=now + timedelta(seconds=index * 2),
+                )
+                for index in range(count)
+            ]
+        )
+        db.flush()
+        risk_repo.evaluate_behavior_rule(
+            db,
+            rule_code=risk_repo.RULE_ONECLICK_DAILY,
+            subject_id=device_id,
+            at=now,
+        )
+    db.commit()
+
+
+def _seed_compare(db, now) -> None:
+    local_now = now.astimezone(risk_repo.CN_TZ).replace(tzinfo=None)
+    accounts = [
+        ("13612343281", "risk-demo-compare-device-a", 1053),
+        ("13511106208", "risk-demo-compare-device-b", 1052),
+    ]
+    for phone, device_id, count in accounts:
+        user = user_repo.upsert_user_for_login(
+            db, phone=phone, register_channel="sms"
+        )
+        rows = []
+        for index in range(count):
+            rows.append(
+                ComparisonRecord(
+                    user_id=user.id,
+                    device_id=device_id,
+                    trace_id=f"{DEMO_PREFIX}compare-{user.id}-{index}",
+                    store_name=("春熙路小吃店", "科技园轻食", "万达广场烤肉")[
+                        index % 3
+                    ],
+                    product_names=("招牌套餐", "鸡胸沙拉", "双人烤肉")[index % 3],
+                    status="failed" if index % 17 == 0 else "success",
+                    information="演示比价记录",
+                    items=[{"name": "演示菜品", "qty": 1}],
+                    comparison_results=[
+                        {
+                            "platform_id": "meituan",
+                            "platform_name": "美团",
+                            "price": 23.8,
+                        },
+                        {
+                            "platform_id": "taobao",
+                            "platform_name": "淘宝",
+                            "price": 21.5,
+                        },
+                        {
+                            "platform_id": "jd",
+                            "platform_name": "京东",
+                            "price": 22.2,
+                        },
+                    ],
+                    skipped_dish_names=[],
+                    saved_amount_cents=230,
+                    raw_payload={"demo": True},
+                    created_at=local_now + timedelta(seconds=index * 2),
+                )
+            )
+        db.add_all(rows)
+        db.commit()
+        risk_repo.sync_compare_incident(db, user_id=user.id, at=local_now)
+
+
+def main() -> None:
+    if settings.APP_ENV != "dev":
+        raise SystemExit("拒绝执行：风控演示数据脚本仅允许 APP_ENV=dev")
+    now = risk_repo.utcnow().replace(minute=10, second=0, microsecond=0)
+    with SessionLocal() as db:
+        _reset_demo(db)
+        _ensure_admin(db)
+        _seed_sms(db, now)
+        _seed_oneclick(db, now)
+        _seed_compare(db, now)
+    print("风控监控演示数据已重建")
+    print(f"后台账号：{ADMIN_USERNAME}")
+    print(f"后台密码：{ADMIN_PASSWORD}")
+    print("期望卡片：短信 3 / 1284；一键登录 2 / 963；比价 2 / 2105")
+
+
+if __name__ == "__main__":
+    main()

scripts/sync_pangle_revenue.py

@@ -1,7 +1,7 @@
-"""每日拉取穿山甲 GroMore 天级收益报表入库(供 admin 广告收益报表的「穿山甲后台收益」)。
+"""每日拉取 GroMore 排序价预估与 ADN Reporting 收益入库(供 admin 广告收益对账)。
 
-GroMore 数据 API 为 T+1:次日穿山甲约 10:00 出数。建议线上每天 ~10:30 由 systemd timer 跑一次
-(默认拉【昨天】);穿山甲对历史数据可能订正,故支持回补近 N 天(幂等 upsert,重跑无害)。
+GroMore 数据 API 为 T+1:次日约 10:00 出初值，第三方 ADN Reporting 最晚约 13:50 更新。
+线上由 systemd timer 在 10:30、14:30 各跑一次;历史数据可能订正,故支持回补近 N 天。
 
 用法:
     python -m scripts.sync_pangle_revenue                 # 拉昨天(北京时间)
@@ -117,7 +117,7 @@ def sync_range(start_date: str, end_date: str) -> None:
         stats = repo.upsert_daily_rows(db, rows)
     total_rev = round(sum(r["revenue_yuan"] for r in rows), 4)
     print(f"✅ 完成:接口 {len(raw)} 行 → 入库 {len(rows)} 行(跳过 {skipped}),"
-          f"新增 {stats['inserted']} / 更新 {stats['updated']};预估收益合计 ¥{total_rev}")
+          f"新增 {stats['inserted']} / 更新 {stats['updated']};排序价预估合计 ¥{total_rev}")
 
 
 def main() -> None:

tests/test_admin_ad_revenue_scope.py

@@ -8,12 +8,15 @@ from sqlalchemy import delete
 from app.admin.repositories import ad_revenue
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
+from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_pangle_revenue import AdPangleDailyRevenue
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
 
 REPORT_DATE = "2040-02-03"
 PLAYBACK_DATE = "2040-02-04"
+DETAIL_DATE = "2040-02-06"
+SOURCE_FALLBACK_DATE = "2040-02-07"
 
 
 def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -> None:
@@ -50,18 +53,22 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="prod", our_code_id="prod-reward",
                 adn="", revenue_yuan=1.5, api_revenue_yuan=1.2, impressions=10,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="prod", our_code_id="prod-demo",
                 adn="", revenue_yuan=8.0, api_revenue_yuan=7.0, impressions=40,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="prod", our_code_id="104098712",
                 adn="", revenue_yuan=2.5, api_revenue_yuan=2.0, impressions=20,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
             AdPangleDailyRevenue(
                 report_date=REPORT_DATE, app_env="test", our_code_id="104127529",
                 adn="", revenue_yuan=9.0, api_revenue_yuan=8.0, impressions=50,
+                synced_at=datetime(2040, 2, 4, 6, 30, tzinfo=UTC),
             ),
         ])
         db.commit()
@@ -86,6 +93,8 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
         assert business["total_revenue_yuan"] == 0.5
         assert business["total_pangle_revenue_yuan"] == 4.0
         assert business["total_pangle_api_revenue_yuan"] == 3.2
+        assert business["pangle_api_revenue_complete"] is True
+        assert business["pangle_latest_synced_at"] is not None
 
         all_codes = ad_revenue.ad_revenue_report(
             db,
@@ -131,7 +140,7 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
         db.close()
 
 
-def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
+def test_reward_video_impression_revenue_is_independent_of_reward_status_and_cap() -> None:
     db = SessionLocal()
     phone = "18800009992"
     sessions = {
@@ -147,13 +156,17 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
 
         for index, (status, session_id) in enumerate(sessions.items(), start=1):
             created_at = datetime(2040, 2, 4, index, tzinfo=UTC)
+            # 发奖状态 capped 的广告故意使用 ¥1000 CPM，验证收入不套用金币侧 ¥500 CPM 封顶。
+            ecpm_raw = "100000" if status == "capped" else "10000"
             db.add(AdEcpmRecord(
                 user_id=user.id,
                 ad_type="reward_video",
                 ad_session_id=session_id,
+                adn=f"adn-{status}",
+                slot_id=f"rit-{status}",
                 app_env="prod",
                 our_code_id="prod-reward",
-                ecpm_raw="10000",
+                ecpm_raw=ecpm_raw,
                 report_date=PLAYBACK_DATE,
                 created_at=created_at,
             ))
@@ -166,7 +179,7 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
                 ad_session_id=session_id,
                 app_env="prod",
                 our_code_id="prod-reward",
-                ecpm_raw="10000",
+                ecpm_raw=ecpm_raw,
                 reward_date=PLAYBACK_DATE,
                 created_at=created_at,
             ))
@@ -185,22 +198,33 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
 
         revenue_by_status = {row["status"]: row["revenue_yuan"] for row in result["items"]}
         assert revenue_by_status == {
-            "closed_early": 0.0,
-            "too_short": 0.0,
-            "capped": 0.1,
+            "closed_early": 0.1,
+            "too_short": 0.1,
+            "capped": 1.0,
             "granted": 0.1,
         }
         assert result["total_impressions"] == 4
-        assert result["total_revenue_yuan"] == 0.2
+        assert result["total_revenue_yuan"] == 1.3
         assert len(result["daily"]) == 1
         assert result["daily"][0]["date"] == PLAYBACK_DATE
         assert result["daily"][0]["impressions"] == 4
-        assert result["daily"][0]["revenue_yuan"] == 0.2
-        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 0.2
+        assert result["daily"][0]["revenue_yuan"] == 1.3
+        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 1.3
         assert result["type_stats"]["reward_video"] == {
             "impressions": 4,
-            "revenue_yuan": 0.2,
+            "revenue_yuan": 1.3,
+            "ecpm_yuan": 325.0,
         }
+        detail_by_status = {
+            row["status"]: row["reward_detail"] for row in result["items"]
+        }
+        assert detail_by_status["granted"]["adn"] == "adn-granted"
+        assert detail_by_status["granted"]["slot_id"] == "rit-granted"
+        # 未进入发奖的记录可保留展示收入，但不能凭空生成奖励因子或占用 LT 累计。
+        for status in ("closed_early", "too_short", "capped"):
+            assert detail_by_status[status]["ecpm_factor"] is None
+            assert detail_by_status[status]["lt_factor_start"] is None
+            assert detail_by_status[status]["lt_index_start"] is None
     finally:
         db.rollback()
         db.execute(delete(AdRewardRecord).where(AdRewardRecord.reward_date == PLAYBACK_DATE))
@@ -208,3 +232,214 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         db.execute(delete(User).where(User.phone == phone))
         db.commit()
         db.close()
+
+
+def test_category_stats_merge_legacy_feed_and_withdrawal_video() -> None:
+    db = SessionLocal()
+    phone = "18800009993"
+    category_date = "2040-02-05"
+    try:
+        user = User(phone=phone, username="29999999993", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            # Draw 经营分类必须包含新 draw 与历史 feed。
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="category-draw",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="feed", ad_session_id="category-feed",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
+            ),
+            # 看视频经营分类必须包含福利与提现两个视频入口。
+            AdEcpmRecord(
+                user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=category_date,
+            date_to=category_date,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+
+        assert result["category_stats"] == {
+            "draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
+            "video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
+def test_feed_reward_detail_keeps_each_record_adn_instead_of_parent_adn() -> None:
+    db = SessionLocal()
+    phone = "18800009994"
+    try:
+        user = User(phone=phone, username="29999999994", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdFeedRewardRecord(
+                client_event_id="detail-adn-pangle",
+                user_id=user.id,
+                reward_date=DETAIL_DATE,
+                duration_seconds=20,
+                unit_count=1,
+                ecpm_raw="12000",
+                adn="pangle",
+                slot_id="rit-pangle",
+                ad_type="draw",
+                feed_scene="coupon",
+                trace_id="detail-adn-trace",
+                app_env="prod",
+                our_code_id="104098712",
+                coin=12,
+                status="granted",
+                created_at=datetime(2040, 2, 6, 1, tzinfo=UTC),
+            ),
+            AdFeedRewardRecord(
+                client_event_id="detail-adn-gdt",
+                user_id=user.id,
+                reward_date=DETAIL_DATE,
+                duration_seconds=20,
+                unit_count=1,
+                ecpm_raw="25000",
+                adn="gdt",
+                slot_id="rit-gdt",
+                ad_type="draw",
+                feed_scene="coupon",
+                trace_id="detail-adn-trace",
+                app_env="prod",
+                our_code_id="104098712",
+                coin=25,
+                status="granted",
+                created_at=datetime(2040, 2, 6, 2, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=DETAIL_DATE,
+            date_to=DETAIL_DATE,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+
+        item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
+        assert item["adn"] is None
+        assert [detail["adn"] for detail in item["sub_rewards"]] == ["pangle", "gdt"]
+        assert [detail["slot_id"] for detail in item["sub_rewards"]] == ["rit-pangle", "rit-gdt"]
+    finally:
+        db.rollback()
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == DETAIL_DATE))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
+def test_feed_reward_source_can_fallback_to_unique_trace_impression_only() -> None:
+    """发奖会话是整场 ID、展示会话是 impressionId 时，只在 trace+eCPM 唯一时回填 ADN。"""
+    db = SessionLocal()
+    phone = "18800009995"
+    try:
+        user = User(phone=phone, username="29999999995", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdFeedRewardRecord(
+                client_event_id="source-fallback-unique", user_id=user.id,
+                reward_date=SOURCE_FALLBACK_DATE, duration_seconds=3, unit_count=0,
+                ecpm_raw="4700", ad_session_id="flow-session", trace_id="source-trace",
+                app_env="prod", our_code_id="104098712", coin=0, status="too_short",
+                ad_type="draw", feed_scene="comparison",
+                created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
+            ),
+            AdFeedRewardRecord(
+                client_event_id="source-fallback-ambiguous", user_id=user.id,
+                reward_date=SOURCE_FALLBACK_DATE, duration_seconds=3, unit_count=0,
+                ecpm_raw="4800", ad_session_id="flow-session", trace_id="source-trace",
+                app_env="prod", our_code_id="104098712", coin=0, status="too_short",
+                ad_type="draw", feed_scene="comparison",
+                created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="impression-unique",
+                trace_id="source-trace", ecpm_raw="4700", adn="baidu", slot_id="rit-baidu",
+                app_env="prod", our_code_id="104098712", report_date=SOURCE_FALLBACK_DATE,
+                created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-a",
+                trace_id="source-trace", ecpm_raw="4800", adn="baidu", slot_id="rit-baidu",
+                app_env="prod", our_code_id="104098712", report_date=SOURCE_FALLBACK_DATE,
+                created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-b",
+                trace_id="source-trace", ecpm_raw="4800", adn="ks", slot_id="rit-ks",
+                app_env="prod", our_code_id="104098712", report_date=SOURCE_FALLBACK_DATE,
+                created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=SOURCE_FALLBACK_DATE,
+            date_to=SOURCE_FALLBACK_DATE,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+        item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
+        details = {detail["ecpm"]: detail for detail in item["sub_rewards"]}
+        assert details["4700"]["adn"] == "baidu"
+        assert details["4700"]["slot_id"] == "rit-baidu"
+        assert details["4800"]["adn"] is None
+        assert details["4800"]["slot_id"] is None
+    finally:
+        db.rollback()
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == SOURCE_FALLBACK_DATE))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == SOURCE_FALLBACK_DATE))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
+def test_pangle_api_day_is_provisional_before_14_beijing_time() -> None:
+    assert ad_revenue._pangle_api_day_complete(
+        REPORT_DATE,
+        {
+            "api_revenue_yuan": 3.2,
+            # D+1 10:30 北京时间。
+            "synced_at": datetime(2040, 2, 4, 2, 30, tzinfo=UTC),
+        },
+    ) is False
+    assert ad_revenue._pangle_api_day_complete(
+        REPORT_DATE,
+        {
+            "api_revenue_yuan": 3.2,
+            # D+1 14:00 北京时间，达到日终判定线。
+            "synced_at": datetime(2040, 2, 4, 6, 0, tzinfo=UTC),
+        },
+    ) is True

tests/test_admin_roles.py

@@ -68,10 +68,17 @@ def test_monitoring_audit_catalog_and_api_permissions(
     ).json()
     monitoring = next(group for group in catalog if group["group"] == "监控审计")
     assert [page["key"] for page in monitoring["pages"]] == [
-        "device-liveness", "analytics-health", "event-logs", "audit-logs",
+        "risk-monitor",
+        "device-liveness",
+        "analytics-health",
+        "event-logs",
+        "audit-logs",
     ]
 
-    # 运营默认只能查设备存活，不能绕过导航直调技术/审计接口。
+    # 运营默认可查风控和设备存活，不能绕过导航直调技术/审计接口。
+    assert admin_client.get(
+        "/admin/api/risk-monitor/summary", headers=_auth(operator_token)
+    ).status_code == 200
     assert admin_client.get(
         "/admin/api/device-liveness/stats", headers=_auth(operator_token)
     ).status_code == 200
@@ -82,8 +89,9 @@ def test_monitoring_audit_catalog_and_api_permissions(
     ):
         assert admin_client.get(path, headers=_auth(operator_token)).status_code == 403
 
-    # 技术角色默认拥有监控审计组全部四项权限。
+    # 技术角色默认拥有监控审计组全部五项权限。
     for path in (
+        "/admin/api/risk-monitor/summary",
         "/admin/api/device-liveness/stats",
         "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
         "/admin/api/event-logs",
@@ -184,8 +192,8 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
     assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
     assert set(roles["tech"]["pages"]) == {
-        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
-        "event-logs", "audit-logs",
+        "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
+        "ad-revenue", "huawei-review", "event-logs", "audit-logs",
     }
 
 

tests/test_derive_best_excludes_short.py

@@ -1,102 +0,0 @@
-"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"。
-
-回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
-会把缺菜店的虚低价当 best_price → 记录页戴"最低"红框 + 虚假省额。
-修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)。
-纯函数, 不碰 DB。
-"""
-from app.repositories.comparison import _derive, _derive_from_results
-from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
-
-
-def test_derive_from_results_excludes_short_ordered_from_best():
-    # jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
-    results = [
-        {"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
-        {"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
-        {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
-    ]
-    platform_results = {
-        "jd_waimai": {"skipped_dish_count": 2},
-        "taobao_flash": {"skipped_dish_count": 0},
-        "meituan": {"is_source": True},
-    }
-    d = _derive_from_results(results, platform_results)
-    assert d["best_platform_id"] == "taobao_flash"
-    assert d["best_price_cents"] == 3850
-    assert d["saved_amount_cents"] == 4200 - 3850   # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
-    assert d["is_source_best"] is False
-
-
-def test_derive_from_results_all_targets_short_falls_back_to_source():
-    # 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
-    d = _derive_from_results(results, platform_results)
-    assert d["best_platform_id"] == "meituan"
-    assert d["is_source_best"] is True
-    assert d["saved_amount_cents"] == 0
-
-
-def test_derive_from_results_no_platform_results_keeps_old_behavior():
-    # 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    d = _derive_from_results(results)
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500
-
-
-def test_derive_from_results_malformed_platform_results_no_crash():
-    # 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    d = _derive_from_results(results, {"jd_waimai": "oops"})
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500
-
-
-def test_derive_pydantic_excludes_short():
-    # _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
-    payload = ComparisonRecordIn(
-        trace_id="t-short-pyd",
-        source_price=42.0,
-        source_platform_id="meituan",
-        comparison_results=[
-            ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
-            ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
-            ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
-        ],
-        platform_results={
-            "jd_waimai": {"skipped_dish_count": 2},
-            "taobao_flash": {"skipped_dish_count": 0},
-        },
-    )
-    d = _derive(payload)
-    assert d["best_platform_id"] == "taobao_flash"
-    assert d["best_price_cents"] == 3850
-    assert d["saved_amount_cents"] == 4200 - 3850
-    assert d["is_source_best"] is False
-
-
-def test_derive_pydantic_malformed_platform_results_no_crash():
-    # _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
-    payload = ComparisonRecordIn(
-        trace_id="t-malformed-pyd",
-        source_price=42.0,
-        comparison_results=[
-            ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
-            ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
-        ],
-        platform_results={"jd_waimai": "oops"},
-    )
-    d = _derive(payload)  # 不抛 AttributeError
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500

tests/test_invite_cash_withdraw.py

@@ -128,8 +128,7 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None:
 
 
 def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
-    """两账户各提各的不串:先提 invite_cash(拒绝结清),再提 cash,各扣各账户。
-    注:一个用户同一时间只能一个活跃提现单(跨账户),故第二笔需先结清第一笔。"""
+    """两账户各提各的不串:提 invite_cash(拒绝→验证退款回原账户),再提 cash,各扣各账户互不串。"""
     _patch_userinfo(monkeypatch, "openid_ic_3")
     token = _login(client, "13800004003")
     _seed_balances(client, token, "13800004003", cash=400, invite_cash=500)
@@ -140,7 +139,7 @@ def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
         json={"amount_cents": 200, "source": "invite_cash"},
         headers=_auth(token),
     )
-    _reject(r1.json()["out_bill_no"])    # 退回 invite_cash + 结清活跃单
+    _reject(r1.json()["out_bill_no"])    # 拒绝退回 invite_cash(验证退款回原账户)
     r2 = client.post(
         "/api/v1/wallet/withdraw",
         # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
@@ -154,6 +153,39 @@ def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     assert cash == 350           # 扣了 cash 50
 
 
+def test_invite_cash_multiple_in_flight_allowed(client, monkeypatch) -> None:
+    """取消「同时仅一单」:invite_cash 无档位上限,已有在途时仍可继续提交,多笔并存各扣 invite_cash。"""
+    _patch_userinfo(monkeypatch, "openid_ic_multi")
+    token = _login(client, "13800004006")
+    _seed_balances(client, token, "13800004006", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r1 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash", "out_bill_no": "billicflight0001"},
+        headers=_auth(token),
+    )
+    assert r1.status_code == 200, r1.text
+    assert r1.json()["status"] == "reviewing"
+
+    r2 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash", "out_bill_no": "billicflight0002"},
+        headers=_auth(token),
+    )
+    assert r2.status_code == 200, r2.text
+    assert r2.json()["status"] == "reviewing"
+
+    # 两笔 invite_cash 在途并存,共扣 400(500→100),金币现金不动
+    cash, invite_cash = _balances(client, token)
+    assert invite_cash == 100 and cash == 0
+    orders = client.get(
+        "/api/v1/wallet/withdraw-orders", params={"source": "invite_cash"}, headers=_auth(token)
+    ).json()["items"]
+    reviewing = [o for o in orders if o["status"] == "reviewing"]
+    assert len(reviewing) == 2
+
+
 def test_invite_me_returns_reward_stats(client) -> None:
     """/invite/me 返回 reward_balance_cents(可提现奖励金)。"""
     token = _login(client, "13800004004")
@@ -170,12 +202,11 @@ def test_withdraw_orders_source_filter(client, monkeypatch) -> None:
     _seed_balances(client, token, "13800004005", cash=400, invite_cash=500)
     client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
 
-    r1 = client.post(
+    client.post(
         "/api/v1/wallet/withdraw",
         json={"amount_cents": 200, "source": "invite_cash"},
         headers=_auth(token),
     )
-    _reject(r1.json()["out_bill_no"])    # 结清,才能提第二笔
     client.post(
         "/api/v1/wallet/withdraw",
         # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)

tests/test_risk_monitor.py

@@ -0,0 +1,541 @@
+"""风控监控：三类规则、明细及处置闭环。"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from uuid import uuid4
+
+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.api.v1 import compare as compare_api
+from app.core.config_schema import (
+    RISK_COMPARE_DAILY_THRESHOLD_KEY,
+    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+    RISK_SMS_HOURLY_THRESHOLD_KEY,
+)
+from app.core.security import issue_token_pair
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.models.device import DeviceLiveness
+from app.models.risk import RiskIncident
+from app.repositories import risk as risk_repo
+from app.repositories import user as user_repo
+
+
+@pytest.fixture(autouse=True)
+def _reset_risk_rule_config():
+    keys = (
+        RISK_SMS_HOURLY_THRESHOLD_KEY,
+        RISK_ONECLICK_DAILY_THRESHOLD_KEY,
+        RISK_COMPARE_DAILY_THRESHOLD_KEY,
+        risk_repo.RISK_RESET_BASELINES_KEY,
+    )
+    with SessionLocal() as db:
+        db.query(AppConfig).filter(AppConfig.key.in_(keys)).delete(
+            synchronize_session=False
+        )
+        db.commit()
+    yield
+    with SessionLocal() as db:
+        db.query(AppConfig).filter(AppConfig.key.in_(keys)).delete(
+            synchronize_session=False
+        )
+        db.commit()
+
+
+def _admin_token() -> str:
+    username = "risk_monitor_admin"
+    with SessionLocal() as db:
+        if admin_repo.get_by_username(db, username) is None:
+            admin_repo.create_admin(
+                db, username=username, password="risk-pass-123", role="super_admin"
+            )
+    with TestClient(admin_app) as client:
+        response = client.post(
+            "/admin/api/auth/login",
+            json={"username": username, "password": "risk-pass-123"},
+        )
+    assert response.status_code == 200
+    return response.json()["access_token"]
+
+
+def _headers() -> dict[str, str]:
+    return {"Authorization": f"Bearer {_admin_token()}"}
+
+
+def _seed_incidents() -> tuple[int, int, int, int, str]:
+    suffix = uuid4().hex[:8]
+    sms_subject = f"risk_sms_{suffix}"
+    oneclick_subject = f"risk_oneclick_{suffix}"
+    now = (risk_repo.utcnow() - timedelta(minutes=5)).replace(microsecond=0)
+    with SessionLocal() as db:
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone=f"139{int(suffix, 16) % 100_000_000:08d}",
+            register_channel="sms",
+        )
+        second_user = user_repo.upsert_user_for_login(
+            db,
+            phone=f"137{int(suffix, 16) % 100_000_000:08d}",
+            register_channel="sms",
+        )
+        db.add(
+            DeviceLiveness(
+                user_id=user.id,
+                device_id=sms_subject,
+                created_at=now - timedelta(days=10),
+            )
+        )
+        for index in range(5):
+            risk_repo.record_behavior_event(
+                db,
+                event_type=risk_repo.EVENT_SMS_SEND,
+                subject_type="device",
+                subject_id=sms_subject,
+                device_id=sms_subject,
+                device_model="OPPO A5",
+                phone=f"13877771{index:03d}",
+                outcome="success",
+                occurred_at=now + timedelta(seconds=index),
+                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
+            )
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_LOGIN,
+            subject_type="device",
+            subject_id=sms_subject,
+            user_id=user.id,
+            device_id=sms_subject,
+            device_model="OPPO A5",
+            phone="13877771000",
+            outcome="success",
+            occurred_at=now + timedelta(seconds=30),
+        )
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_LOGIN,
+            subject_type="device",
+            subject_id=sms_subject,
+            user_id=second_user.id,
+            device_id=sms_subject,
+            device_model="OPPO A5",
+            phone="13877771001",
+            outcome="success",
+            occurred_at=now + timedelta(seconds=31),
+        )
+
+        for index in range(21):
+            risk_repo.record_behavior_event(
+                db,
+                event_type=risk_repo.EVENT_ONECLICK_LOGIN,
+                subject_type="device",
+                subject_id=oneclick_subject,
+                user_id=second_user.id if index == 20 else user.id,
+                device_id=oneclick_subject,
+                device_model="OPPO Reno",
+                phone=user.phone,
+                outcome="success" if index % 3 else "failed",
+                reason="mock provider failure" if index % 3 == 0 else None,
+                occurred_at=now + timedelta(seconds=index),
+                evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
+            )
+
+        local_now = now.astimezone(risk_repo.CN_TZ).replace(tzinfo=None)
+        for index in range(100):
+            db.add(
+                ComparisonRecord(
+                    user_id=user.id,
+                    device_id="risk_compare_device",
+                    trace_id=f"risk-monitor-{suffix}-{index}",
+                    store_name="测试门店",
+                    product_names="测试菜品",
+                    status="success",
+                    items=[{"name": "测试菜品", "qty": 1}],
+                    comparison_results=[
+                        {"platform_id": "meituan", "platform_name": "美团", "price": 12.5},
+                        {"platform_id": "taobao", "platform_name": "淘宝", "price": 11.8},
+                    ],
+                    skipped_dish_names=[],
+                    saved_amount_cents=70,
+                    created_at=local_now + timedelta(seconds=index),
+                )
+            )
+        db.commit()
+        risk_repo.sync_compare_incident(db, user_id=user.id, at=local_now)
+
+        incidents = {
+            incident.rule_code: incident.id
+            for incident in db.query(RiskIncident)
+            .filter(
+                RiskIncident.subject_id.in_(
+                    (sms_subject, oneclick_subject, str(user.id))
+                )
+            )
+            .all()
+        }
+        return (
+            incidents[risk_repo.RULE_SMS_HOURLY],
+            incidents[risk_repo.RULE_ONECLICK_DAILY],
+            incidents[risk_repo.RULE_COMPARE_DAILY],
+            user.id,
+            sms_subject,
+        )
+
+
+def test_risk_monitor_summary_lists_and_details() -> None:
+    sms_id, oneclick_id, compare_id, _, _ = _seed_incidents()
+    headers = _headers()
+    with TestClient(admin_app) as client:
+        summary = client.get("/admin/api/risk-monitor/summary", headers=headers)
+        assert summary.status_code == 200
+        cards = {card["kind"]: card for card in summary.json()["cards"]}
+        assert cards["sms"]["alert_subject_count"] >= 1
+        assert cards["sms"]["today_total"] >= 5
+        assert cards["oneclick"]["alert_subject_count"] >= 1
+        assert cards["oneclick"]["today_total"] >= 20
+        assert cards["compare"]["alert_subject_count"] >= 1
+        assert cards["compare"]["today_total"] >= 100
+
+        sms_list = client.get("/admin/api/risk-monitor/incidents/sms", headers=headers)
+        assert sms_list.status_code == 200
+        sms_item = next(
+            item for item in sms_list.json()["items"] if item["incident_id"] == sms_id
+        )
+        assert sms_item["device_model"] == "OPPO A5"
+        assert sms_item["event_count"] == 5
+        assert sms_item["first_used_at"].startswith(
+            (risk_repo.utcnow() - timedelta(days=10)).date().isoformat()
+        )
+
+        sms_detail = client.get(
+            f"/admin/api/risk-monitor/incidents/sms/{sms_id}/details",
+            headers=headers,
+        )
+        assert sms_detail.status_code == 200
+        assert sms_detail.json()["total"] == 5
+        assert sms_detail.json()["distinct_accounts"] == 2
+        assert any(item["verified"] for item in sms_detail.json()["items"])
+
+        oneclick_detail = client.get(
+            f"/admin/api/risk-monitor/incidents/oneclick/{oneclick_id}/details?limit=1",
+            headers=headers,
+        )
+        assert oneclick_detail.status_code == 200
+        assert oneclick_detail.json()["total"] == 21
+        assert oneclick_detail.json()["distinct_accounts"] == 2
+
+        compare_detail = client.get(
+            f"/admin/api/risk-monitor/incidents/compare/{compare_id}/details",
+            headers=headers,
+        )
+        assert compare_detail.status_code == 200
+        assert compare_detail.json()["total"] == 100
+        assert compare_detail.json()["items"][0]["prices"]["美团"] == 12.5
+
+
+def test_risk_monitor_ignore_and_block_are_enforced(client: TestClient) -> None:
+    sms_id, oneclick_id, compare_id, user_id, sms_subject = _seed_incidents()
+    headers = _headers()
+    with TestClient(admin_app) as admin_client:
+        ignored = admin_client.post(
+            f"/admin/api/risk-monitor/incidents/{oneclick_id}/ignore",
+            json={"reason": "测试确认正常"},
+            headers=headers,
+        )
+        assert ignored.status_code == 200
+        assert ignored.json()["status"] == "ignored"
+
+        blocked_device = admin_client.post(
+            f"/admin/api/risk-monitor/incidents/{sms_id}/block",
+            json={"reason": "测试设备异常"},
+            headers=headers,
+        )
+        assert blocked_device.status_code == 200
+        assert blocked_device.json()["restriction_id"]
+
+        blocked_user = admin_client.post(
+            f"/admin/api/risk-monitor/incidents/{compare_id}/block",
+            json={"reason": "测试账号异常"},
+            headers=headers,
+        )
+        assert blocked_user.status_code == 200
+        device_restriction_id = blocked_device.json()["restriction_id"]
+        user_restriction_id = blocked_user.json()["restriction_id"]
+
+        blocked_sms = admin_client.get(
+            "/admin/api/risk-monitor/incidents/sms?status=blocked",
+            headers=headers,
+        )
+        assert blocked_sms.status_code == 200
+        assert blocked_sms.json()["items"][0]["restriction_id"] == device_restriction_id
+        assert blocked_sms.json()["items"][0]["restricted"] is True
+
+        blocked_compare = admin_client.get(
+            "/admin/api/risk-monitor/incidents/compare?status=blocked",
+            headers=headers,
+        )
+        assert blocked_compare.status_code == 200
+        assert blocked_compare.json()["items"][0]["restriction_id"] == user_restriction_id
+        assert blocked_compare.json()["items"][0]["restricted"] is True
+
+    denied_sms = client.post(
+        "/api/v1/auth/sms/send",
+        json={
+            "phone": "13877779999",
+            "device_id": sms_subject,
+            "device_model": "OPPO A5",
+        },
+    )
+    assert denied_sms.status_code == 403
+    assert denied_sms.json()["detail"] == "当前设备环境异常，暂无法发送验证码"
+
+    tokens = issue_token_pair(user_id)
+    denied_compare = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": "risk-blocked-new-trace", "device_id": "risk_compare_device"},
+        headers={"Authorization": f"Bearer {tokens['access_token']}"},
+    )
+    assert denied_compare.status_code == 403
+    assert denied_compare.json()["detail"] == "账号存在异常，该功能暂不可用"
+
+    denied_compare_step = client.post(
+        "/api/v1/intent/recognize",
+        json={},
+        headers={"Authorization": f"Bearer {tokens['access_token']}"},
+    )
+    assert denied_compare_step.status_code == 403
+    assert denied_compare_step.json()["detail"] == "账号存在异常，该功能暂不可用"
+
+    with TestClient(admin_app) as admin_client:
+        assert (
+            admin_client.post(
+                f"/admin/api/risk-monitor/restrictions/{device_restriction_id}/revoke",
+                headers=headers,
+            ).status_code
+            == 200
+        )
+        assert (
+            admin_client.post(
+                f"/admin/api/risk-monitor/restrictions/{user_restriction_id}/revoke",
+                headers=headers,
+            ).status_code
+            == 200
+        )
+
+    allowed_sms = client.post(
+        "/api/v1/auth/sms/send",
+        json={
+            "phone": "13877779998",
+            "device_id": sms_subject,
+            "device_model": "OPPO A5",
+        },
+    )
+    assert allowed_sms.status_code == 200
+
+    with SessionLocal() as db:
+        assert not risk_repo.is_restricted(
+            db,
+            subject_type="user",
+            subject_id=str(user_id),
+            scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
+        )
+        assert db.get(RiskIncident, compare_id).status == "resolved"
+
+
+def test_admin_can_reset_all_alerts_and_sms_restarts_from_zero() -> None:
+    sms_id, oneclick_id, compare_id, _, sms_subject = _seed_incidents()
+    headers = _headers()
+
+    with TestClient(admin_app) as client:
+        reset = client.post("/admin/api/risk-monitor/reset", headers=headers)
+        assert reset.status_code == 200
+        payload = reset.json()
+        assert payload["reset_incident_count"] >= 3
+        assert payload["reset_counts"]["sms"] >= 1
+        assert payload["reset_counts"]["oneclick"] >= 1
+        assert payload["reset_counts"]["compare"] >= 1
+
+        summary = client.get("/admin/api/risk-monitor/summary", headers=headers)
+        assert summary.status_code == 200
+        cards = {card["kind"]: card for card in summary.json()["cards"]}
+        assert all(card["alert_subject_count"] == 0 for card in cards.values())
+        # 重置只清待处理报警，不删除今日事实总量。
+        assert cards["sms"]["today_total"] >= 5
+        assert cards["oneclick"]["today_total"] >= 20
+        assert cards["compare"]["today_total"] >= 100
+
+    reset_at = datetime.fromisoformat(payload["reset_at"])
+    with SessionLocal() as db:
+        for incident_id in (sms_id, oneclick_id, compare_id):
+            incident = db.get(RiskIncident, incident_id)
+            assert incident.status == "resolved"
+            assert incident.action_reason == risk_repo.MANUAL_RESET_REASON
+
+        for index in range(4):
+            risk_repo.record_behavior_event(
+                db,
+                event_type=risk_repo.EVENT_SMS_SEND,
+                subject_type="device",
+                subject_id=sms_subject,
+                device_id=sms_subject,
+                phone=f"13688880{index:03d}",
+                outcome="success",
+                occurred_at=reset_at + timedelta(seconds=index + 1),
+                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
+            )
+        assert db.get(RiskIncident, sms_id).status == "resolved"
+
+        risk_repo.record_behavior_event(
+            db,
+            event_type=risk_repo.EVENT_SMS_SEND,
+            subject_type="device",
+            subject_id=sms_subject,
+            device_id=sms_subject,
+            phone="13688880999",
+            outcome="success",
+            occurred_at=reset_at + timedelta(seconds=5),
+            evaluate_rule=risk_repo.RULE_SMS_HOURLY,
+        )
+        incident = db.get(RiskIncident, sms_id)
+        assert incident.status == "open"
+        assert incident.event_count == 5
+        assert incident.window_start == reset_at.replace(tzinfo=None)
+
+
+def test_real_compare_harvest_creates_incident_at_threshold() -> None:
+    suffix = uuid4().hex[:8]
+    local_now = risk_repo.utcnow().astimezone(risk_repo.CN_TZ).replace(tzinfo=None)
+    with SessionLocal() as db:
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone=f"136{int(suffix, 16) % 100_000_000:08d}",
+            register_channel="sms",
+        )
+        for index in range(99):
+            db.add(
+                ComparisonRecord(
+                    user_id=user.id,
+                    device_id="real-harvest-device",
+                    trace_id=f"risk-real-harvest-{suffix}-{index}",
+                    status="running",
+                    created_at=local_now + timedelta(milliseconds=index),
+                )
+            )
+        db.commit()
+        user_id = user.id
+
+    compare_api._harvest_running_blocking(
+        f"risk-real-harvest-{suffix}-99",
+        user_id,
+        "food",
+        "real-harvest-device",
+        {"model": "OPPO A5"},
+        None,
+    )
+
+    with SessionLocal() as db:
+        incident = db.query(RiskIncident).filter(
+            RiskIncident.rule_code == risk_repo.RULE_COMPARE_DAILY,
+            RiskIncident.subject_id == str(user_id),
+            RiskIncident.window_key == local_now.strftime("%Y-%m-%d"),
+        ).one()
+        assert incident.event_count == 100
+
+
+def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
+    suffix = uuid4().hex[:8]
+    subject = f"risk_dynamic_sms_{suffix}"
+    now = risk_repo.utcnow().replace(microsecond=0)
+    with SessionLocal() as db:
+        for index in range(3):
+            risk_repo.record_behavior_event(
+                db,
+                event_type=risk_repo.EVENT_SMS_SEND,
+                subject_type="device",
+                subject_id=subject,
+                device_id=subject,
+                device_model="动态规则测试机",
+                phone=f"13800001{index:03d}",
+                outcome="success",
+                occurred_at=now + timedelta(seconds=index),
+            )
+
+    headers = _headers()
+    with TestClient(admin_app) as client:
+        defaults = client.get("/admin/api/risk-monitor/rules", headers=headers)
+        assert defaults.status_code == 200
+        assert defaults.json() == {
+            "sms_hourly_threshold": 5,
+            "oneclick_daily_threshold": 20,
+            "compare_daily_threshold": 100,
+        }
+
+        lowered = client.patch(
+            "/admin/api/risk-monitor/rules",
+            headers=headers,
+            json={
+                "sms_hourly_threshold": 3,
+                "oneclick_daily_threshold": 20,
+                "compare_daily_threshold": 100,
+            },
+        )
+        assert lowered.status_code == 200
+        with SessionLocal() as db:
+            incident = db.query(RiskIncident).filter(
+                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+                RiskIncident.subject_id == subject,
+            ).one()
+            assert incident.status == "open"
+            assert incident.event_count == 3
+
+        raised = client.patch(
+            "/admin/api/risk-monitor/rules",
+            headers=headers,
+            json={
+                "sms_hourly_threshold": 4,
+                "oneclick_daily_threshold": 20,
+                "compare_daily_threshold": 100,
+            },
+        )
+        assert raised.status_code == 200
+        with SessionLocal() as db:
+            incident = db.query(RiskIncident).filter(
+                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+                RiskIncident.subject_id == subject,
+            ).one()
+            assert incident.status == "resolved"
+            assert incident.action_reason == risk_repo.AUTO_RESOLVED_REASON
+
+        reopened = client.patch(
+            "/admin/api/risk-monitor/rules",
+            headers=headers,
+            json={
+                "sms_hourly_threshold": 3,
+                "oneclick_daily_threshold": 20,
+                "compare_daily_threshold": 100,
+            },
+        )
+        assert reopened.status_code == 200
+        with SessionLocal() as db:
+            incident = db.query(RiskIncident).filter(
+                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+                RiskIncident.subject_id == subject,
+            ).one()
+            assert incident.status == "open"
+            assert incident.triggered_at == (
+                now + timedelta(seconds=2)
+            ).replace(tzinfo=None)
+
+        invalid = client.patch(
+            "/admin/api/risk-monitor/rules",
+            headers=headers,
+            json={
+                "sms_hourly_threshold": 21,
+                "oneclick_daily_threshold": 20,
+                "compare_daily_threshold": 100,
+            },
+        )
+        assert invalid.status_code == 422

tests/test_test_account.py

@@ -190,6 +190,42 @@ def test_exempt_from_device_ip_send_rate_limit(client, enabled, monkeypatch) ->
         )
 
 
+def test_send_is_audit_only_and_does_not_trigger_risk_alert(client, enabled) -> None:
+    """测试号不真发短信：可留审计流水，但不能算正式短信下发或触发设备风控。"""
+    from app.db.session import SessionLocal
+    from app.models.risk import BehaviorEvent, RiskIncident
+    from app.repositories import risk as risk_repo
+
+    device_id = "qa-test-account-device"
+    for _ in range(6):
+        response = client.post(
+            "/api/v1/auth/sms/send",
+            json={"phone": TEST_PHONE, "device_id": device_id},
+        )
+        assert response.status_code == 200
+
+    with SessionLocal() as db:
+        events = (
+            db.query(BehaviorEvent)
+            .filter(
+                BehaviorEvent.event_type == risk_repo.EVENT_SMS_SEND,
+                BehaviorEvent.subject_id == device_id,
+            )
+            .all()
+        )
+        assert len(events) == 6
+        assert {event.outcome for event in events} == {"test"}
+        assert (
+            db.query(RiskIncident)
+            .filter(
+                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
+                RiskIncident.subject_id == device_id,
+            )
+            .count()
+            == 0
+        )
+
+
 # ============================ 计数单元逻辑 ============================
 
 def test_quota_resets_across_day(enabled, monkeypatch) -> None:

tests/test_withdraw.py

@@ -268,6 +268,61 @@ def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
     assert sum(1 for t in r.json()["items"] if t["biz_type"] == "withdraw") == 1
 
 
+def test_withdraw_multiple_in_flight_allowed(client, monkeypatch) -> None:
+    """取消「同时仅一单」:已有在途(reviewing)时,不同 out_bill_no 可继续提交,两单并存。"""
+    _patch_userinfo(monkeypatch, "openid_multi_inflight")
+    token = _login(client, "13800002020")
+    _seed_cash(client, token, "13800002020", 100)  # 够两笔 0.5
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r1 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billmulti00000001"},
+        headers=_auth(token),
+    )
+    assert r1.status_code == 200, r1.text
+    assert r1.json()["status"] == "reviewing"
+
+    r2 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billmulti00000002"},
+        headers=_auth(token),
+    )
+    assert r2.status_code == 200, r2.text
+    assert r2.json()["status"] == "reviewing"
+
+    # 两张在途单并存
+    r = client.get("/api/v1/wallet/withdraw-orders", headers=_auth(token))
+    reviewing = [o for o in r.json()["items"] if o["status"] == "reviewing"]
+    assert len(reviewing) == 2, r.text
+    # 余额扣两次:100-50-50=0
+    r = client.get("/api/v1/wallet/account", headers=_auth(token))
+    assert r.json()["cash_balance_cents"] == 0
+
+
+def test_withdraw_second_blocked_only_by_insufficient_cash(client, monkeypatch) -> None:
+    """并行放开后第二笔仅受余额约束:余额不足返 409「现金余额不足」,而非旧的「已有提现」拦截。"""
+    _patch_userinfo(monkeypatch, "openid_multi_insuff")
+    token = _login(client, "13800002021")
+    _seed_cash(client, token, "13800002021", 50)  # 仅够一笔 0.5
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r1 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billinsuff0000001"},
+        headers=_auth(token),
+    )
+    assert r1.status_code == 200, r1.text
+
+    r2 = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50, "out_bill_no": "billinsuff0000002"},
+        headers=_auth(token),
+    )
+    assert r2.status_code == 409, r2.text
+    assert "现金余额不足" in r2.json()["detail"]
+
+
 def test_withdraw_ambiguous_timeout_then_success_no_refund(client, monkeypatch) -> None:
     """#3 转账调用超时(异常),但查单确认已 SUCCESS → 不退款,单置 success。"""
     monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_amb", "nickname": None, "avatar_url": None, "raw": {}})