test(withdraw): 补 invite_cash 并发在途用例 + 清理过期注释

审查发现:取消「一人一单」后,create_withdraw 的档位闸注释与
test_invite_cash_withdraw 中若干注释仍引用已删除的「在途互斥」。
- create_withdraw 注释去掉「在途互斥」措辞
- test_two_accounts_withdraw_independent 文案修正;source_filter
  用例移除已非必需的 reject(断言不变)
- 新增 test_invite_cash_multiple_in_flight_allowed:两笔 invite_cash
  在途并存、各扣邀请奖励金、不串金币现金

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

alembic/versions/drop_withdraw_active_unique_index.py

@@ -1,6 +1,6 @@
 """drop withdraw active-order partial unique index (allow multiple in-flight withdrawals)
 
-Revision ID: drop_withdraw_active_uniq_idx
+Revision ID: drop_withdraw_active_unique_index
 Revises: d8dd2106e438
 Create Date: 2026-07-24 00:00:00.000000
 
@@ -12,7 +12,7 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision: str = "drop_withdraw_active_uniq_idx"
+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

alembic/versions/risk_monitor_generic_tables.py

@@ -1,210 +0,0 @@
-"""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")

alembic/versions/user_manual_risk_fields.py

@@ -1,76 +0,0 @@
-"""add manual high-risk flag and note to user
-
-Revision ID: user_manual_risk_fields
-Revises: risk_monitor_generic
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "user_manual_risk_fields"
-down_revision: str | None = "risk_monitor_generic"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-_INVITE_WITHDRAW_PAGE = "invite-withdraws"
-
-
-def _finance_role() -> sa.TableClause:
-    return sa.table(
-        "admin_role",
-        sa.column("name", sa.String),
-        sa.column("pages", _JSON),
-    )
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("user") as batch_op:
-        batch_op.add_column(
-            sa.Column(
-                "is_high_risk",
-                sa.Boolean(),
-                server_default=sa.false(),
-                nullable=False,
-            )
-        )
-        batch_op.add_column(sa.Column("high_risk_note", sa.Text(), nullable=True))
-        batch_op.create_index("ix_user_is_high_risk", ["is_high_risk"], unique=False)
-    role = _finance_role()
-    conn = op.get_bind()
-    row = conn.execute(
-        sa.select(role.c.pages).where(role.c.name == "finance")
-    ).scalar_one_or_none()
-    if row is not None and _INVITE_WITHDRAW_PAGE not in (row or []):
-        conn.execute(
-            role.update()
-            .where(role.c.name == "finance")
-            .values(pages=[*(row or []), _INVITE_WITHDRAW_PAGE])
-        )
-
-
-def downgrade() -> None:
-    role = _finance_role()
-    conn = op.get_bind()
-    row = conn.execute(
-        sa.select(role.c.pages).where(role.c.name == "finance")
-    ).scalar_one_or_none()
-    if row is not None:
-        conn.execute(
-            role.update()
-            .where(role.c.name == "finance")
-            .values(
-                pages=[
-                    page for page in (row or []) if page != _INVITE_WITHDRAW_PAGE
-                ]
-            )
-        )
-    with op.batch_alter_table("user") as batch_op:
-        batch_op.drop_index("ix_user_is_high_risk")
-        batch_op.drop_column("high_risk_note")
-        batch_op.drop_column("is_high_risk")

app/admin/main.py

@@ -17,7 +17,6 @@ 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
@@ -26,6 +25,8 @@ 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
@@ -33,9 +34,7 @@ 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
@@ -102,7 +101,6 @@ 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

@@ -23,8 +23,7 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "cps", "label": "CPS收益"},
     ]},
     {"group": "奖励审核", "pages": [
-        {"key": "invite-withdraws", "label": "邀请提现审核"},
-        {"key": "withdraws", "label": "其他提现审核"},
+        {"key": "withdraws", "label": "提现审核"},
         {"key": "price-reports", "label": "低价审核"},
         {"key": "feedbacks", "label": "用户反馈"},
     ]},
@@ -35,7 +34,6 @@ 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": "埋点日志"},
@@ -57,13 +55,13 @@ BUILTIN_ROLES: list[dict] = [
     {"name": SUPER_ADMIN_ROLE, "label": "管理员", "pages": []},
     {"name": "operator", "label": "运营", "pages": [
         "dashboard", "coupon-data", "ad-revenue-report", "comparison-records",
-        "cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
+        "cps", "device-liveness", "price-reports", "feedbacks", "huawei-review",
     ]},
     {"name": "finance", "label": "财务", "pages": [
-        "dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws",
+        "dashboard", "ad-revenue-report", "cps", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
-        "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
+        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
         "event-logs", "audit-logs",
     ]},
 ]

app/admin/repositories/mutations.py

@@ -44,25 +44,6 @@ def set_user_debug_trace(
     return user
 
 
-def set_user_risk(
-    db: Session,
-    user: User,
-    *,
-    is_high_risk: bool,
-    note: str | None,
-    commit: bool = True,
-) -> User:
-    """设置人工风险结论和备注，支持与审计日志共用同一事务。"""
-    user.is_high_risk = is_high_risk
-    user.high_risk_note = note.strip() if is_high_risk and note else None
-    if commit:
-        db.commit()
-        db.refresh(user)
-    else:
-        db.flush()
-    return user
-
-
 def update_feedback_status(
     db: Session, feedback: Feedback, *, status: str, commit: bool = True
 ) -> Feedback:

app/admin/repositories/queries.py

@@ -22,15 +22,14 @@ from app.models.comparison import ComparisonRecord
 from app.models.coupon_state import CouponPromptEngagement
 from app.models.device import DeviceLiveness
 from app.models.feedback import Feedback
-from app.models.invite import InviteRelation
 from app.models.onboarding import OnboardingCompletion
 from app.models.price_report import PriceReport
-from app.models.savings import SavingsRecord
 from app.models.user import User
 from app.models.wallet import (
     CashTransaction,
     CoinAccount,
     CoinTransaction,
+    InviteCashTransaction,
     WithdrawOrder,
 )
 from app.repositories import activity, ad_ecpm
@@ -711,7 +710,16 @@ def list_all_withdraw_orders(
     elif quick_filter == "today":
         stmt = stmt.where(WithdrawOrder.created_at >= today_start)
     elif quick_filter == "high_risk":
-        stmt = stmt.where(User.is_high_risk.is_(True))
+        stmt = stmt.where(
+            or_(
+                WithdrawOrder.user_name.is_(None),
+                WithdrawOrder.user_name == "",
+                User.status != "active",
+                User.created_at >= now - timedelta(hours=24),
+                WithdrawOrder.status.in_(("failed", "rejected")),
+                WithdrawOrder.fail_reason.is_not(None),
+            )
+        )
 
     sort_cols = {
         "id": WithdrawOrder.id,
@@ -737,9 +745,7 @@ def _as_utc(value: datetime) -> datetime:
     return value.astimezone(timezone.utc)
 
 
-def withdraw_list_enrichment(
-    db: Session, user_ids: list[int], *, source: str | None = None
-) -> dict[int, dict]:
+def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict]:
     """批量富化提现单列表:按本页 user_id 取 手机号/昵称 + 各自累计成功提现金额(分)。
 
     两条聚合查询搞定(避免逐行 N+1)。累计口径与 get_user_overview 的 withdraw_success_cents
@@ -749,32 +755,25 @@ def withdraw_list_enrichment(
     uniq = list(set(user_ids))
     if not uniq:
         return {}
-    success_stmt = select(
-        WithdrawOrder.user_id,
-        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
-    ).where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
-    if source:
-        success_stmt = success_stmt.where(WithdrawOrder.source == source)
-    success_rows = db.execute(success_stmt.group_by(WithdrawOrder.user_id)).all()
+    success_rows = db.execute(
+        select(
+            WithdrawOrder.user_id,
+            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+        )
+        .where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
+        .group_by(WithdrawOrder.user_id)
+    ).all()
     success_map = {uid: int(total) for uid, total in success_rows}
     users = db.execute(
-        select(
-            User.id,
-            User.phone,
-            User.nickname,
-            User.is_high_risk,
-            User.high_risk_note,
-        ).where(User.id.in_(uniq))
+        select(User.id, User.phone, User.nickname).where(User.id.in_(uniq))
     ).all()
     return {
         uid: {
             "phone": phone,
             "nickname": nickname,
-            "is_high_risk": is_high_risk,
-            "high_risk_note": high_risk_note,
             "cumulative_success_cents": success_map.get(uid, 0),
         }
-        for uid, phone, nickname, is_high_risk, high_risk_note in users
+        for uid, phone, nickname in users
     }
 
 
@@ -880,16 +879,15 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder
     ).scalar_one_or_none()
 
 
-def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
+def withdraw_summary(db: Session) -> dict:
     """提现审核台顶部统计。金额单位:分。"""
-    summary_stmt = select(
-        WithdrawOrder.status,
-        func.count(WithdrawOrder.id),
-        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
-    )
-    if source:
-        summary_stmt = summary_stmt.where(WithdrawOrder.source == source)
-    rows = db.execute(summary_stmt.group_by(WithdrawOrder.status)).all()
+    rows = db.execute(
+        select(
+            WithdrawOrder.status,
+            func.count(WithdrawOrder.id),
+            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+        ).group_by(WithdrawOrder.status)
+    ).all()
     by_status = {
         status: {"count": int(count), "amount_cents": int(amount_cents)}
         for status, count, amount_cents in rows
@@ -902,23 +900,19 @@ def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
     )
 
     def _today_count(status: str) -> int:
-        stmt = select(func.count(WithdrawOrder.id)).where(
-            WithdrawOrder.status == status,
+        return db.execute(
+            select(func.count(WithdrawOrder.id)).where(
+                WithdrawOrder.status == status,
+                WithdrawOrder.updated_at >= today_start,
+            )
+        ).scalar_one()
+
+    today_success_amount = db.execute(
+        select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
+            WithdrawOrder.status == "success",
             WithdrawOrder.updated_at >= today_start,
         )
-        if source:
-            stmt = stmt.where(WithdrawOrder.source == source)
-        return db.execute(stmt).scalar_one()
-
-    today_amount_stmt = select(
-        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)
-    ).where(
-        WithdrawOrder.status == "success",
-        WithdrawOrder.updated_at >= today_start,
-    )
-    if source:
-        today_amount_stmt = today_amount_stmt.where(WithdrawOrder.source == source)
-    today_success_amount = db.execute(today_amount_stmt).scalar_one()
+    ).scalar_one()
 
     return {
         "reviewing_count": by_status.get("reviewing", {}).get("count", 0),
@@ -931,117 +925,6 @@ def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
     }
 
 
-def invite_overview(
-    db: Session,
-    inviter_user_id: int,
-    *,
-    date_from: datetime | None = None,
-    date_to: datetime | None = None,
-) -> dict:
-    """邀请提现详情：邀请关系统计及受邀用户首次比价/首单信息。
-
-    首次记录按自增 id 取最早一条，批量查询避免按受邀用户逐行查询。
-    邀请成功口径为完成首次比价并已发邀请奖励(compare_reward_granted)。
-    """
-    relation_stmt = (
-        select(InviteRelation, User)
-        .join(User, User.id == InviteRelation.invitee_user_id)
-        .where(InviteRelation.inviter_user_id == inviter_user_id)
-        .order_by(InviteRelation.created_at.asc(), InviteRelation.id.asc())
-    )
-    if date_from is not None:
-        relation_stmt = relation_stmt.where(User.created_at >= _as_utc_naive(date_from))
-    if date_to is not None:
-        relation_stmt = relation_stmt.where(User.created_at <= _as_utc_naive(date_to))
-    relation_rows = db.execute(relation_stmt).all()
-    invitee_ids = [relation.invitee_user_id for relation, _ in relation_rows]
-    if not invitee_ids:
-        return {"invite_total": 0, "invite_success_total": 0, "items": []}
-
-    first_compare_ids = list(
-        db.execute(
-            select(func.min(ComparisonRecord.id))
-            .where(
-                ComparisonRecord.user_id.in_(invitee_ids),
-                ComparisonRecord.status == "success",
-            )
-            .group_by(ComparisonRecord.user_id)
-        ).scalars()
-    )
-    comparisons = (
-        db.execute(
-            select(
-                ComparisonRecord.user_id,
-                ComparisonRecord.store_name,
-                ComparisonRecord.product_names,
-            ).where(ComparisonRecord.id.in_(first_compare_ids))
-        ).all()
-        if first_compare_ids
-        else []
-    )
-    comparison_map = {
-        user_id: (store_name, product_names)
-        for user_id, store_name, product_names in comparisons
-    }
-
-    first_order_ids = list(
-        db.execute(
-            select(func.min(SavingsRecord.id))
-            .where(
-                SavingsRecord.user_id.in_(invitee_ids),
-                SavingsRecord.source == "compare",
-            )
-            .group_by(SavingsRecord.user_id)
-        ).scalars()
-    )
-    orders = (
-        db.execute(
-            select(
-                SavingsRecord.user_id,
-                SavingsRecord.shop_name,
-                SavingsRecord.title,
-                SavingsRecord.dishes,
-                SavingsRecord.order_amount_cents,
-            ).where(SavingsRecord.id.in_(first_order_ids))
-        ).all()
-        if first_order_ids
-        else []
-    )
-    order_map = {
-        user_id: (
-            shop_name,
-            title or "、".join(dishes or []),
-            order_amount_cents,
-        )
-        for user_id, shop_name, title, dishes, order_amount_cents in orders
-    }
-
-    items = []
-    for relation, user in relation_rows:
-        compare_store, compare_products = comparison_map.get(user.id, (None, None))
-        order_store, order_products, order_amount = order_map.get(
-            user.id, (None, None, None)
-        )
-        items.append(
-            {
-                "user_id": user.id,
-                "phone": user.phone,
-                "registered_at": user.created_at,
-                "invite_success": bool(relation.compare_reward_granted),
-                "first_compare_store": compare_store,
-                "first_compare_products": compare_products,
-                "first_order_store": order_store,
-                "first_order_products": order_products,
-                "first_order_amount_cents": order_amount,
-            }
-        )
-    return {
-        "invite_total": len(items),
-        "invite_success_total": sum(1 for item in items if item["invite_success"]),
-        "items": items,
-    }
-
-
 def list_withdraw_audit_logs(
     db: Session, out_bill_no: str, *, limit: int = 20
 ) -> list[AdminAuditLog]:
@@ -1062,6 +945,8 @@ def withdraw_risk_flags(
     cash_balance_cents: int,
 ) -> tuple[list[str], int]:
     flags: list[str] = []
+    if not order.user_name:
+        flags.append("缺少提现实名")
     if user and user.status != "active":
         flags.append(f"账号状态:{user.status}")
     if user and user.created_at:
@@ -1084,6 +969,125 @@ def withdraw_risk_flags(
     return flags, score
 
 
+def _check_withdraw_ledger_side(
+    orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str
+) -> dict:
+    """对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。
+
+    orders 已按 source 过滤到本账;txns 是本账流水表里 withdraw_biz/refund_biz 两类流水。
+    规则:每单发起应有一条扣款流水(ref_id=out_bill_no);失败/拒绝单应有且仅一条退款流水;
+    非退款终态不应出现退款流水。四个计数全为 0 即本账自洽。
+    """
+    withdraw_refs = {txn.ref_id for txn in txns if txn.biz_type == withdraw_biz}
+    refund_counts: dict[str, int] = {}
+    for txn in txns:
+        if txn.biz_type == refund_biz and txn.ref_id:
+            refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
+
+    missing_withdraw = 0
+    missing_refund = 0
+    refund_on_non_terminal = 0
+    for order in orders:
+        if order.out_bill_no not in withdraw_refs:
+            missing_withdraw += 1
+        has_refund = refund_counts.get(order.out_bill_no, 0) > 0
+        if order.status in {"failed", "rejected"} and not has_refund:
+            missing_refund += 1
+        if has_refund and order.status not in {"failed", "rejected"}:
+            refund_on_non_terminal += 1
+
+    return {
+        "missing_withdraw": missing_withdraw,
+        "missing_refund": missing_refund,
+        "duplicate_refund": sum(1 for count in refund_counts.values() if count > 1),
+        "refund_on_non_terminal": refund_on_non_terminal,
+    }
+
+
+def withdraw_ledger_check(db: Session) -> dict:
+    """现金账本校验:两本物理隔离的账各自对账(产品红线:coin_cash / invite_cash 不串)。
+
+    普通现金:CoinAccount.cash_balance_cents ↔ cash_transaction(withdraw/withdraw_refund);
+    邀请奖励金:CoinAccount.invite_cash_balance_cents ↔ invite_cash_transaction
+              (invite_withdraw/invite_withdraw_refund)。
+    提现单按 source 分流到对应账核对——邀请提现的流水写在 invite_cash_transaction 表,
+    绝不能拿去和普通现金流水比(否则每笔邀请提现单都会被误报「缺扣款/缺退款流水」)。
+    分流口径与 create_withdraw 一致:仅 source==invite_cash 走邀请账,其余(含历史空值)归普通现金。
+    """
+    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
+    coin_orders = [o for o in orders if o.source != "invite_cash"]
+    invite_orders = [o for o in orders if o.source == "invite_cash"]
+
+    # —— 普通现金账(coin_cash) ——
+    cash_balance_total = int(
+        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
+    )
+    cash_txn_total = int(
+        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
+    )
+    cash_txns = list(
+        db.execute(
+            select(CashTransaction).where(
+                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    coin = _check_withdraw_ledger_side(
+        coin_orders, cash_txns, withdraw_biz="withdraw", refund_biz="withdraw_refund"
+    )
+    cash_diff = cash_balance_total - cash_txn_total
+
+    # —— 邀请奖励金账(invite_cash,独立账户 + 独立流水表) ——
+    invite_balance_total = int(
+        db.execute(
+            select(func.coalesce(func.sum(CoinAccount.invite_cash_balance_cents), 0))
+        ).scalar_one()
+    )
+    invite_txn_total = int(
+        db.execute(
+            select(func.coalesce(func.sum(InviteCashTransaction.amount_cents), 0))
+        ).scalar_one()
+    )
+    invite_txns = list(
+        db.execute(
+            select(InviteCashTransaction).where(
+                InviteCashTransaction.biz_type.in_(("invite_withdraw", "invite_withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    invite = _check_withdraw_ledger_side(
+        invite_orders, invite_txns,
+        withdraw_biz="invite_withdraw", refund_biz="invite_withdraw_refund",
+    )
+    invite_diff = invite_balance_total - invite_txn_total
+
+    ok = (
+        cash_diff == 0
+        and invite_diff == 0
+        and all(v == 0 for v in coin.values())
+        and all(v == 0 for v in invite.values())
+    )
+    return {
+        "ok": ok,
+        # 普通现金账(coin_cash:金币兑换的现金)
+        "cash_balance_total_cents": cash_balance_total,
+        "cash_transaction_total_cents": cash_txn_total,
+        "balance_diff_cents": cash_diff,
+        "missing_withdraw_txn_count": coin["missing_withdraw"],
+        "missing_refund_txn_count": coin["missing_refund"],
+        "duplicate_refund_txn_count": coin["duplicate_refund"],
+        "refund_txn_on_non_terminal_count": coin["refund_on_non_terminal"],
+        # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)
+        "invite_cash_balance_total_cents": invite_balance_total,
+        "invite_cash_transaction_total_cents": invite_txn_total,
+        "invite_balance_diff_cents": invite_diff,
+        "invite_missing_withdraw_txn_count": invite["missing_withdraw"],
+        "invite_missing_refund_txn_count": invite["missing_refund"],
+        "invite_duplicate_refund_txn_count": invite["duplicate_refund"],
+        "invite_refund_txn_on_non_terminal_count": invite["refund_on_non_terminal"],
+    }
+
+
 def get_user_overview(db: Session, user_id: int) -> dict | None:
     """用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
     user = db.get(User, user_id)
@@ -1143,7 +1147,6 @@ def user_reward_stats(
     *,
     date_from: datetime | None = None,
     date_to: datetime | None = None,
-    withdraw_source: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
@@ -1151,35 +1154,22 @@ def user_reward_stats(
     平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
-    withdraw_source_conds = (
-        [WithdrawOrder.source == withdraw_source] if withdraw_source else []
-    )
     wd_success = db.execute(
         select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
             WithdrawOrder.user_id == user_id,
             WithdrawOrder.status == "success",
-            *withdraw_source_conds,
             *_window_conds(WithdrawOrder.created_at, date_from, date_to),
         )
     ).scalar_one()
     wd_total = db.execute(
         select(func.count(WithdrawOrder.id)).where(
             WithdrawOrder.user_id == user_id,
-            *withdraw_source_conds,
             *_window_conds(WithdrawOrder.created_at, date_from, date_to),
         )
     ).scalar_one()
 
     acc = db.get(CoinAccount, user_id)  # 现金余额:当前快照,不随窗口
-    cash_balance = (
-        (
-            acc.invite_cash_balance_cents
-            if withdraw_source == "invite_cash"
-            else acc.cash_balance_cents
-        )
-        if acc
-        else 0
-    )
+    cash_balance = acc.cash_balance_cents if acc else 0
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。

app/admin/repositories/risk_monitor.py

@@ -1,545 +0,0 @@
-"""风控监控读模型：聚合通用行为流水、风险事件与比价记录。"""
-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/config.py

@@ -30,12 +30,6 @@ 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

@@ -1,331 +0,0 @@
-"""风控监控：统计、风险事件列表/详情及忽略、封禁、解封。"""
-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/routers/users.py

@@ -16,7 +16,6 @@ from app.admin.schemas.user import (
     GrantCashRequest,
     GrantCoinsRequest,
     SetDebugTraceRequest,
-    SetUserRiskRequest,
     SetUserStatusRequest,
     UserCoinRecord,
     UserRewardStats,
@@ -85,21 +84,12 @@ def get_user_reward_stats(
     db: AdminDb,
     date_from: Annotated[datetime | None, Query()] = None,
     date_to: Annotated[datetime | None, Query()] = None,
-    withdraw_source: Annotated[
-        str | None, Query(pattern="^(coin_cash|invite_cash)$")
-    ] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
     if user_repo.get_user_by_id(db, user_id) is None:
         raise HTTPException(status_code=404, detail="用户不存在")
     return UserRewardStats(
-        **queries.user_reward_stats(
-            db,
-            user_id,
-            date_from=date_from,
-            date_to=date_to,
-            withdraw_source=withdraw_source,
-        )
+        **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)
     )
 
 
@@ -148,50 +138,6 @@ def set_user_status(
     return OkResponse()
 
 
-@router.post("/{user_id}/risk", response_model=OkResponse, summary="设置人工高风险标记与备注")
-def set_user_risk(
-    user_id: int,
-    body: SetUserRiskRequest,
-    request: Request,
-    admin: Annotated[
-        AdminUser, Depends(require_role("operator", "finance"))
-    ],
-    db: AdminDb,
-) -> OkResponse:
-    user = user_repo.get_user_by_id(db, user_id)
-    if user is None:
-        raise HTTPException(status_code=404, detail="用户不存在")
-    before = {
-        "is_high_risk": user.is_high_risk,
-        "high_risk_note": user.high_risk_note,
-    }
-    mutations.set_user_risk(
-        db,
-        user,
-        is_high_risk=body.is_high_risk,
-        note=body.note,
-        commit=False,
-    )
-    write_audit(
-        db,
-        admin,
-        action="user.risk.set",
-        target_type="user",
-        target_id=user_id,
-        detail={
-            "before": before,
-            "after": {
-                "is_high_risk": user.is_high_risk,
-                "high_risk_note": user.high_risk_note,
-            },
-        },
-        ip=get_client_ip(request),
-        commit=False,
-    )
-    db.commit()
-    return OkResponse()
-
-
 @router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限")
 def set_user_debug_trace(
     user_id: int,

app/admin/routers/withdraw.py

@@ -1,4 +1,4 @@
-"""admin 提现:列表(读)+ 单笔/批量查单与审核操作(写,带审计)。
+"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
 
 提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
 这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
@@ -14,16 +14,17 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.repositories import queries
-from app.admin.schemas.admin import AdminAuditLogOut
 from app.admin.schemas.common import CursorPage
+from app.admin.schemas.admin import AdminAuditLogOut
 from app.admin.schemas.wallet import (
     CashTxnOut,
-    InviteOverviewOut,
-    WithdrawBulkItemResult,
+    ReconcileResult,
     WithdrawBulkRejectRequest,
     WithdrawBulkRequest,
     WithdrawBulkResult,
+    WithdrawBulkItemResult,
     WithdrawDetailOut,
+    WithdrawLedgerCheckOut,
     WithdrawListItemOut,
     WithdrawOrderOut,
     WithdrawRejectRequest,
@@ -82,16 +83,8 @@ def list_withdraws(
         cursor=cursor,
     )
     # 联表带出手机号/昵称 + 累计成功提现(本页 user_id 批量富化,2 条聚合查询,无 N+1)。
-    enrichment = queries.withdraw_list_enrichment(
-        db, [o.user_id for o in items], source=source
-    )
-    _empty = {
-        "phone": None,
-        "nickname": None,
-        "is_high_risk": False,
-        "high_risk_note": None,
-        "cumulative_success_cents": 0,
-    }
+    enrichment = queries.withdraw_list_enrichment(db, [o.user_id for o in items])
+    _empty = {"phone": None, "nickname": None, "cumulative_success_cents": 0}
     out_items = [
         WithdrawListItemOut.model_validate(o).model_copy(
             update=enrichment.get(o.user_id, _empty)
@@ -102,13 +95,8 @@ def list_withdraws(
 
 
 @router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
-def withdraws_summary(
-    db: AdminDb,
-    source: Annotated[
-        str | None, Query(pattern="^(coin_cash|invite_cash)$")
-    ] = None,
-) -> WithdrawSummaryOut:
-    return WithdrawSummaryOut(**queries.withdraw_summary(db, source=source))
+def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
+    return WithdrawSummaryOut(**queries.withdraw_summary(db))
 
 
 @router.get(
@@ -165,13 +153,13 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
     )
 
 
+@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
+def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
+    return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
+
+
 @router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
-def withdraw_detail(
-    out_bill_no: str,
-    db: AdminDb,
-    date_from: Annotated[datetime | None, Query()] = None,
-    date_to: Annotated[datetime | None, Query()] = None,
-) -> WithdrawDetailOut:
+def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
     order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
     if order is None:
         raise HTTPException(status_code=404, detail="提现单不存在")
@@ -184,8 +172,6 @@ def withdraw_detail(
             id=user.id,
             phone=user.phone,
             nickname=user.nickname,
-            is_high_risk=user.is_high_risk,
-            high_risk_note=user.high_risk_note,
             status=user.status,
             wechat_nickname=user.wechat_nickname,
             wechat_avatar_url=user.wechat_avatar_url,
@@ -209,36 +195,37 @@ def withdraw_detail(
         recent_withdraws,
         overview["cash_balance_cents"] if overview else 0,
     )
-    detail_enrichment = queries.withdraw_list_enrichment(
-        db, [order.user_id], source=order.source
-    ).get(order.user_id, {})
 
     return WithdrawDetailOut(
         order=WithdrawOrderOut.model_validate(order),
         user=user_snapshot,
-        cumulative_success_cents=int(
-            detail_enrichment.get("cumulative_success_cents", 0)
-        ),
         risk_flags=risk_flags,
         risk_score=risk_score,
         recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
         recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
         audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
-        invite_overview=(
-            InviteOverviewOut(
-                **queries.invite_overview(
-                    db,
-                    order.user_id,
-                    date_from=date_from,
-                    date_to=date_to,
-                )
-            )
-            if order.source == "invite_cash"
-            else None
-        ),
     )
 
 
+# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
+@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
+def reconcile(
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+    older_than_minutes: Annotated[int, Query(ge=0)] = 15,
+) -> ReconcileResult:
+    try:
+        result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
+    except wxpay.WxPayNotConfiguredError as e:
+        raise HTTPException(status_code=503, detail="微信支付未配置") from e
+    write_audit(
+        db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None,
+        detail=result, ip=get_client_ip(request), commit=True,
+    )
+    return ReconcileResult(**result)
+
+
 def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
     success = sum(1 for item in items if item.ok)
     return WithdrawBulkResult(

app/admin/schemas/risk_monitor.py

@@ -1,104 +0,0 @@
-"""风控监控后台接口 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/admin/schemas/user.py

@@ -16,8 +16,6 @@ class AdminUserListItem(BaseModel):
     register_channel: str
     status: str
     debug_trace_enabled: bool = False
-    is_high_risk: bool = False
-    high_risk_note: str | None = None
     wechat_openid: str | None = None
     wechat_nickname: str | None = None
     created_at: datetime
@@ -120,19 +118,3 @@ class SetUserStatusRequest(BaseModel):
 
 class SetDebugTraceRequest(BaseModel):
     enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")
-
-
-class SetUserRiskRequest(BaseModel):
-    is_high_risk: bool = Field(..., description="是否标记为高风险用户")
-    note: str | None = Field(
-        None,
-        max_length=500,
-        description="高风险备注；标记高风险时必填，解除后清空",
-    )
-
-    @field_validator("note")
-    @classmethod
-    def validate_note(cls, value: str | None, info):
-        if info.data.get("is_high_risk") and not (value or "").strip():
-            raise ValueError("标记高风险时必须填写原因")
-        return value.strip() if value and value.strip() else None

app/admin/schemas/wallet.py

@@ -60,8 +60,6 @@ class WithdrawListItemOut(WithdrawOrderOut):
 
     phone: str | None = None
     nickname: str | None = None
-    is_high_risk: bool = False
-    high_risk_note: str | None = None
     cumulative_success_cents: int = 0  # 累计成功提现 = SUM(amount_cents) WHERE status='success'
 
 
@@ -79,8 +77,6 @@ class WithdrawUserSnapshot(BaseModel):
     id: int
     phone: str
     nickname: str | None = None
-    is_high_risk: bool = False
-    high_risk_note: str | None = None
     status: str
     wechat_nickname: str | None = None
     wechat_avatar_url: str | None = None
@@ -91,34 +87,19 @@ class WithdrawUserSnapshot(BaseModel):
     withdraw_success_cents: int
 
 
-class InviteeDetailOut(BaseModel):
-    user_id: int
-    phone: str
-    registered_at: datetime
-    invite_success: bool
-    first_compare_store: str | None = None
-    first_compare_products: str | None = None
-    first_order_store: str | None = None
-    first_order_products: str | None = None
-    first_order_amount_cents: int | None = None
-
-
-class InviteOverviewOut(BaseModel):
-    invite_total: int
-    invite_success_total: int
-    items: list[InviteeDetailOut]
-
-
 class WithdrawDetailOut(BaseModel):
     order: WithdrawOrderOut
     user: WithdrawUserSnapshot | None = None
-    cumulative_success_cents: int = 0
     risk_flags: list[str]
     risk_score: int
     recent_withdraws: list[WithdrawOrderOut]
     recent_cash_transactions: list[CashTxnOut]
     audit_logs: list[AdminAuditLogOut]
-    invite_overview: InviteOverviewOut | None = None
+
+
+class ReconcileResult(BaseModel):
+    checked: int
+    resolved: int
 
 
 class WithdrawBulkRequest(BaseModel):
@@ -147,6 +128,26 @@ class WithdrawBulkResult(BaseModel):
     items: list[WithdrawBulkItemResult]
 
 
+class WithdrawLedgerCheckOut(BaseModel):
+    ok: bool
+    # 普通现金账(coin_cash:金币兑换的现金)
+    cash_balance_total_cents: int
+    cash_transaction_total_cents: int
+    balance_diff_cents: int
+    missing_withdraw_txn_count: int
+    missing_refund_txn_count: int
+    duplicate_refund_txn_count: int
+    refund_txn_on_non_terminal_count: int
+    # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)。默认 0 向后兼容。
+    invite_cash_balance_total_cents: int = 0
+    invite_cash_transaction_total_cents: int = 0
+    invite_balance_diff_cents: int = 0
+    invite_missing_withdraw_txn_count: int = 0
+    invite_missing_refund_txn_count: int = 0
+    invite_duplicate_refund_txn_count: int = 0
+    invite_refund_txn_on_non_terminal_count: int = 0
+
+
 class WxpayHealthCheckOut(BaseModel):
     ok: bool
     wxpay_configured: bool

app/api/v1/auth.py

@@ -37,7 +37,6 @@ 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,
@@ -71,18 +70,6 @@ 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:
@@ -100,15 +87,7 @@ def _login_response(
 # ===================== 极光一键登录 =====================
 
 @router.post("/jverify-login", response_model=TokenWithUser, summary="极光一键登录")
-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="当前设备环境异常，暂无法登录")
+def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     logger.info(
         "jverify_login operator=%s token_len=%d",
         req.operator or "-",
@@ -118,37 +97,10 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
     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")
 
@@ -164,34 +116,12 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
     response_model=SmsSendResponse,
     summary="发送短信验证码",
 )
-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="当前设备环境异常，暂无法发送验证码")
+def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
     # 测试账号:不真发短信(号码非真实手机号,真发会失败/浪费),直接放行让客户端进入填码界面。
     # 真正的"免验证码"在 /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,每小时 / 每天两道闸,**均只按成功发码计数**。
@@ -209,18 +139,6 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
     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(被单号冷却挡下的重发走不到这里,故不占额度)
@@ -228,20 +146,6 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
 
     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)
 
 
@@ -251,7 +155,6 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
     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 需在一小时内反复登录联调)。
@@ -259,19 +162,6 @@ 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))
@@ -293,33 +183,9 @@ 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,30 +28,19 @@ import httpx
 from fastapi import APIRouter, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
-from app.api.deps import DbSession, OptionalUser
+from app.api.deps import 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 写法)。
@@ -63,14 +52,10 @@ def _harvest_running_blocking(
     device_id: str | None, device_info: dict | None, trace_url: str | None,
 ) -> None:
     with SessionLocal() as db:
-        rec = crud_compare.harvest_running(
+        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},
@@ -87,10 +72,6 @@ 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,
@@ -218,33 +199,25 @@ async def _forward(
 
 
 @router.post("/intent/recognize", summary="外卖比价 Phase 1 意图识别 (透传 + 建 running 行)")
-async def intent_recognize(
-    request: Request, user: OptionalUser, db: DbSession
-) -> dict[str, Any]:
-    _ensure_compare_allowed(user, db)
+async def intent_recognize(request: Request, user: OptionalUser) -> dict[str, Any]:
     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, db: DbSession) -> dict[str, Any]:
-    _ensure_compare_allowed(user, db)
+async def intent_step(request: Request, user: OptionalUser) -> dict[str, Any]:
     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, db: DbSession
-) -> dict[str, Any]:
-    _ensure_compare_allowed(user, db)
+async def intent_precoupon_step(request: Request, user: OptionalUser) -> dict[str, Any]:
     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, db: DbSession) -> dict[str, Any]:
-    _ensure_compare_allowed(user, db)
+async def price_step(request: Request, user: OptionalUser) -> dict[str, Any]:
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
     # (单平台中途 done 被 pricebot 改写成 wait+continue=true,不会命中这里,同 coupon 语义。)
@@ -264,10 +237,7 @@ async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dic
 
 
 @router.post("/trace/epilogue", summary="比价结果页尾声帧 (透传到 pricebot, 用户视角截图入 trace)")
-async def trace_epilogue(
-    request: Request, user: OptionalUser, db: DbSession
-) -> dict[str, Any]:
-    _ensure_compare_allowed(user, db)
+async def trace_epilogue(request: Request, user: OptionalUser) -> dict[str, Any]:
     # App 收到 done、渲染完结果页后,把自己页面的截图(base64)传给 pricebot 存进 trace
     # 目录并触发重传 —— trace 里补上"用户实际看到的汇总页"(步骤帧只有目标 App 画面)。
     # body ~几百 KB(截图 base64), 纯透传壳(harvest_first_frame=False:不建行/不落库,
@@ -280,10 +250,7 @@ async def trace_epilogue(
 
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
-async def trace_finalize(
-    request: Request, user: OptionalUser, db: DbSession
-) -> dict[str, Any]:
-    _ensure_compare_allowed(user, db)
+async def trace_finalize(request: Request, user: OptionalUser) -> dict[str, Any]:
     # 用户终止 / 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,7 +19,6 @@ 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,
@@ -48,15 +47,8 @@ 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:
-        rec, used = crud_compare.reserve_daily_start(
+        _, used = crud_compare.reserve_daily_start(
             db,
             user_id=user.id,
             trace_id=payload.trace_id,
@@ -73,8 +65,6 @@ 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,
@@ -93,13 +83,6 @@ 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,7 +11,6 @@ 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
 
@@ -37,13 +36,6 @@ 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,9 +20,8 @@ from app.core.rewards import (
     WITHDRAW_MAX_CENTS,
     WITHDRAW_MIN_CENTS,
 )
-from app.models.user import User
-from app.repositories import risk as risk_repo
 from app.repositories import wallet as crud_wallet
+from app.models.user import User
 from app.schemas.welfare import (
     BindWechatRequest,
     BindWechatResultOut,
@@ -205,13 +204,6 @@ 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")

app/core/config_schema.py

@@ -11,10 +11,6 @@ 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 按此过滤)。用于把已下线/已改由
@@ -110,36 +106,4 @@ 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/models/__init__.py

@@ -13,7 +13,13 @@ 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,
@@ -21,12 +27,6 @@ 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,12 +39,11 @@ 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/risk.py

@@ -1,163 +0,0 @@
-"""通用行为事件、风险事件与主体限制。
-
-这三张表不是“风控监控页面专用表”：
-
-- ``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/user.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-from sqlalchemy import Boolean, DateTime, Integer, String, Text, false, func
+from sqlalchemy import Boolean, DateTime, Integer, String, false, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -62,13 +62,6 @@ class User(Base):
         Boolean, nullable=False, default=False, server_default=false()
     )
 
-    # 运营人工风险标记。与自动风控分值分开：这里表达人工复核结论，并保留可编辑备注，
-    # 供邀请提现、其他提现和用户管理三个页面统一展示。
-    is_high_risk: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=False, server_default=false(), index=True
-    )
-    high_risk_note: Mapped[str | None] = mapped_column(Text, nullable=True)
-
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False
     )

app/repositories/risk.py

@@ -1,569 +0,0 @@
-"""通用行为事件、风险规则与主体限制仓库。"""
-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

@@ -6,7 +6,6 @@
 """
 from __future__ import annotations
 
-import logging
 import re
 import unicodedata
 import uuid
@@ -21,6 +20,7 @@ from app.core.config import settings
 from app.core.rewards import COIN_PER_CENT, coins_to_cents
 from app.integrations import wxpay
 from app.models.user import User
+from app.repositories.user import apply_wechat_display_identity
 from app.models.wallet import (
     CashTransaction,
     CoinAccount,
@@ -29,55 +29,12 @@ from app.models.wallet import (
     WechatTransferAuthorization,
     WithdrawOrder,
 )
-from app.repositories.user import apply_wechat_display_identity
 from app.services import notification_events
 
-logger = logging.getLogger(__name__)
-
 # 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
 _WX_STATE_SUCCESS = "SUCCESS"
 _WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
 _WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM"  # 用户还没在微信确认页确认
-_WX_FAIL_REASON_LABELS = {
-    "ACCOUNT_FROZEN": "用户微信账户被冻结",
-    "ACCOUNT_NOT_EXIST": "用户微信账户不存在",
-    "BANK_CARD_ACCOUNT_ABNORMAL": "用户银行卡已销户、冻结、作废或挂失",
-    "BANK_CARD_BANK_INFO_WRONG": "用户登记的银行或分支行信息有误",
-    "BANK_CARD_CARD_INFO_WRONG": "用户银行卡户名或卡号有误",
-    "BANK_CARD_COLLECTIONS_ABOVE_QUOTA": "用户银行卡收款达到限额",
-    "BANK_CARD_PARAM_ERROR": "用户收款银行卡信息错误",
-    "BANK_CARD_STATUS_ABNORMAL": "用户银行卡状态异常",
-    "BLOCK_B2C_USERLIMITAMOUNT_BSRULE_MONTH": "用户本月转账收款已达限额",
-    "BLOCK_B2C_USERLIMITAMOUNT_MONTH": "用户账户存在风险，本月收款受限",
-    "DAY_RECEIVED_COUNT_EXCEED": "用户当日收款次数已达上限",
-    "DAY_RECEIVED_QUOTA_EXCEED": "用户当日收款额度已达上限",
-    "EXCEEDED_ESTIMATED_AMOUNT": "转账金额超过预约金额范围",
-    "ID_CARD_NOT_CORRECT": "收款人身份证校验不通过",
-    "MCH_CANCEL": "商户已撤销付款",
-    "MERCHANT_REJECT": "商户转账验密人已驳回",
-    "MERCHANT_NOT_CONFIRM": "商户转账验密人超时未确认",
-    "NAME_NOT_CORRECT": "收款人姓名校验不通过",
-    "OPENID_INVALID": "用户 OpenID 无效或不属于当前 AppID",
-    "OTHER_FAIL_REASON_TYPE": "微信返回其他失败原因",
-    "OVERDUE_CLOSE": "超过微信系统重试期，订单自动关闭",
-    "PAYEE_ACCOUNT_ABNORMAL": "用户微信账户收款异常",
-    "PAYER_ACCOUNT_ABNORMAL": "商户账户付款受限",
-    "PRODUCT_AUTH_CHECK_FAIL": "商户未开通转账权限或权限已冻结",
-    "REALNAME_ACCOUNT_RECEIVED_QUOTA_EXCEED": "用户微信实名账户收款受限",
-    "REAL_NAME_CHECK_FAIL": "用户未完成微信实名认证",
-    "RECEIVE_ACCOUNT_NOT_CONFIGURE": "商户未配置收款用户列表",
-    "RESERVATION_INFO_NOT_MATCH": "转账信息与预约信息不一致",
-    "RESERVATION_SCENE_NOT_MATCH": "转账场景与预约场景不一致",
-    "RESERVATION_STATE_INVALID": "预约转账单状态异常",
-    "TRANSFER_QUOTA_EXCEED": "用户单笔收款额度已达上限",
-    "TRANSFER_REMARK_SET_FAIL": "微信转账备注设置失败",
-    "TRANSFER_RISK": "该笔转账存在风险，已被微信拦截",
-    "TRANSFER_SCENE_INVALID": "商户未获取当前转账场景",
-    "TRANSFER_SCENE_UNAVAILABLE": "当前转账场景暂不可用",
-    "RELATED_ORDER_TRANSFER_AMOUNT_EXCEED": "关联订单累计付款金额超过上限",
-    "RELATED_ORDER_TRANSFER_COUNT_EXCEED": "关联订单累计付款次数超过上限",
-    "BUDGET_NOT_ENOUGH": "商户预算资金不足",
-}
 # 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
 # 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提
 # (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。
@@ -522,19 +479,6 @@ def _try_deduct_cash(db: Session, user_id: int, amount_cents: int, source: str =
 
 def _add_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> int:
     """原子增加指定账户余额(退款用),返回加后余额。source 决定退回哪个账户(两账户隔离)。"""
-    # Normal withdrawals always create the account before deducting funds. This
-    # fallback covers legacy rows, hand-written fixtures, and broken migrations:
-    # a missing balance snapshot must not make a legitimate refund fail with 500.
-    if db.get(CoinAccount, user_id) is None:
-        logger.error(
-            "withdraw refund found missing coin_account; recreating empty account: "
-            "user_id=%s source=%s amount_cents=%s",
-            user_id,
-            source,
-            amount_cents,
-        )
-        get_or_create_account(db, user_id, commit=False)
-
     col = _balance_col(source)
     db.execute(
         update(CoinAccount)
@@ -638,32 +582,6 @@ def _wx_not_found(result: dict) -> bool:
     return "NOT_FOUND" in str(code)
 
 
-def _wechat_api_error_reason(data: object) -> str:
-    """Format a non-200 WeChat API response for operator display."""
-    if not isinstance(data, dict):
-        return f"微信发起转账失败：{data}"
-    code = str(data.get("code") or "").strip()
-    message = str(data.get("message") or "").strip()
-    if code and message:
-        return f"微信发起转账失败：{message}（{code}）"
-    return f"微信发起转账失败：{message or code or '未知错误'}"
-
-
-def _wechat_terminal_failure_reason(data: dict, state: str) -> str:
-    """Translate WeChat query ``fail_reason`` while preserving unknown codes."""
-    code = str(data.get("fail_reason") or "").strip()
-    if code:
-        label = _WX_FAIL_REASON_LABELS.get(code)
-        if label:
-            return f"微信转账失败：{label}（{code}）"
-        return f"微信转账失败：{code}"
-    if state == "CANCELLED":
-        return "微信转账已撤销（CANCELLED）"
-    if state == "CLOSED":
-        return "微信转账已关闭（CLOSED）"
-    return f"微信转账失败（状态：{state or 'FAIL'}）"
-
-
 def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> None:
     """转账调用结果不明(超时/异常/非200)时,**先查单再决定**,绝不盲目退款(防退款后又到账)。
       - 微信查到 SUCCESS        → 钱已出,置 success,不退款
@@ -689,17 +607,13 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
 
     state = q["data"].get("state", "")
     order.wechat_state = state
-    order.transfer_bill_no = q["data"].get("transfer_bill_no") or order.transfer_bill_no
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
+        order.transfer_bill_no = q["data"].get("transfer_bill_no")
         db.commit()
         notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
-        _refund_withdraw(
-            db,
-            order,
-            reason=_wechat_terminal_failure_reason(q["data"], state),
-        )
+        _refund_withdraw(db, order, reason=reason)
     else:
         order.package_info = q["data"].get("package_info") or order.package_info
         db.commit()
@@ -1053,14 +967,6 @@ def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> Wit
     order.package_info = data.get("package_info")  # 免确认转账无此字段(None);确认模式带它供拉确认页
     if data.get("state") == _WX_STATE_SUCCESS:
         order.status = "success"
-    elif data.get("state") in _WX_STATE_FAILED:
-        _refund_withdraw(
-            db,
-            order,
-            reason=_wechat_terminal_failure_reason(data, str(data.get("state") or "")),
-        )
-        db.refresh(order)
-        return order
     db.commit()
     db.refresh(order)
     if order.status == "success":  # 免确认转账直接到账 → PRD #3 提现到账
@@ -1105,11 +1011,7 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
             return order
         if result["status_code"] != 200:
             # 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态)
-            _settle_after_ambiguous(
-                db,
-                order,
-                reason=_wechat_api_error_reason(result["data"]),
-            )
+            _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
             # 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权
             _refresh_active_auth(db, order.user_id)
             db.refresh(order)
@@ -1137,11 +1039,7 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
         return order
 
     if result["status_code"] != 200:
-        _settle_after_ambiguous(
-            db,
-            order,
-            reason=_wechat_api_error_reason(result["data"]),
-        )
+        _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
         db.refresh(order)
         return order
 
@@ -1198,24 +1096,19 @@ def refresh_withdraw_status(
     ).scalar_one_or_none()
     if order is None:
         raise WithdrawOrderNotFound
-    if order.status not in {"pending", "failed"}:
-        return order
-    enrich_failed_order = order.status == "failed"
+    if order.status != "pending":
+        return order  # 已终态,不再查
 
     try:
         result = wxpay.query_transfer(out_bill_no)
     except wxpay.WxPayNotConfiguredError:
         raise
     except Exception as exc:  # noqa: BLE001 - 查单失败不能把运营后台打成 500
-        if enrich_failed_order:
-            return order
         order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256]
         db.commit()
         db.refresh(order)
         return order
     if result["status_code"] != 200:
-        if enrich_failed_order:
-            return order
         if _wx_not_found(result):
             # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全
             _refund_withdraw(db, order, reason="微信无此单,已退回")
@@ -1224,32 +1117,12 @@ def refresh_withdraw_status(
 
     state = result["data"].get("state", "")
     order.wechat_state = state
-    order.transfer_bill_no = (
-        result["data"].get("transfer_bill_no") or order.transfer_bill_no
-    )
-    if enrich_failed_order:
-        if state in _WX_STATE_FAILED:
-            order.fail_reason = _wechat_terminal_failure_reason(
-                result["data"], state
-            )[:256]
-        elif state == _WX_STATE_SUCCESS:
-            order.fail_reason = (
-                "资金状态异常：本地已退款，但微信查单显示已到账，请人工核查"
-            )
-        db.commit()
-        db.refresh(order)
-        return order
-
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
         db.commit()
         notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
-        _refund_withdraw(
-            db,
-            order,
-            reason=_wechat_terminal_failure_reason(result["data"], state),
-        )
+        _refund_withdraw(db, order, reason=f"微信转账状态 {state}")
     elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed:
         # 用户从确认页回来了却仍未确认 → 视为放弃:撤销微信单(防事后确认导致重复打款)后退款。
         # 撤单失败(可能已被确认进 ACCEPTED 的竞态)则保持 pending,等下次查询。

app/schemas/auth.py

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

scripts/seed_risk_monitor_demo.py

@@ -1,274 +0,0 @@
-"""为本地“风控监控”页面灌入一组可重复的完整演示数据。
-
-仅允许在 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()

tests/test_admin_read.py

@@ -12,9 +12,6 @@ from app.admin.repositories import admin_user as admin_repo
 from app.db.session import SessionLocal, engine
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
-from app.models.invite import InviteRelation
-from app.models.savings import SavingsRecord
-from app.models.user import User
 from app.models.wallet import CashTransaction, WithdrawOrder
 from app.repositories import user as user_repo
 from app.repositories import wallet as wallet_repo
@@ -165,57 +162,6 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     assert records.status_code == 200, records.text
 
 
-def test_user_reward_stats_can_scope_withdrawals_by_account(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    uid = _seed_user_with_data("13800000023")
-    db = SessionLocal()
-    try:
-        account = wallet_repo.get_or_create_account(db, uid)
-        account.cash_balance_cents = 123
-        account.invite_cash_balance_cents = 456
-        db.add_all(
-            [
-                WithdrawOrder(
-                    user_id=uid,
-                    out_bill_no="rewardstatsinvite00000001",
-                    amount_cents=250,
-                    source="invite_cash",
-                    status="success",
-                ),
-                WithdrawOrder(
-                    user_id=uid,
-                    out_bill_no="rewardstatsinvite00000002",
-                    amount_cents=50,
-                    source="invite_cash",
-                    status="reviewing",
-                ),
-            ]
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    coin = admin_client.get(
-        f"/admin/api/users/{uid}/reward-stats",
-        params={"withdraw_source": "coin_cash"},
-        headers=_auth(admin_token),
-    )
-    invite = admin_client.get(
-        f"/admin/api/users/{uid}/reward-stats",
-        params={"withdraw_source": "invite_cash"},
-        headers=_auth(admin_token),
-    )
-    assert coin.status_code == 200, coin.text
-    assert invite.status_code == 200, invite.text
-    assert coin.json()["withdraw_success_cents"] == 100
-    assert coin.json()["withdraw_total"] == 1
-    assert coin.json()["cash_balance_cents"] == 123
-    assert invite.json()["withdraw_success_cents"] == 250
-    assert invite.json()["withdraw_total"] == 2
-    assert invite.json()["cash_balance_cents"] == 456
-
-
 def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
     _seed_user_with_data("13800000003")
     r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
@@ -361,19 +307,10 @@ def test_withdraw_list_exposes_source_and_filters(
     db = SessionLocal()
     try:
         uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id
-        # 直接从 ORM 查询，验证列表富化返回人工风险结论。
-        user = db.get(User, uid)
-        assert user is not None
-        user.is_high_risk = True
-        user.high_risk_note = "邀请行为异常"
         db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001",
                              amount_cents=100, status="success", source="coin_cash"))
         db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001",
                              amount_cents=200, status="success", source="invite_cash"))
-        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview",
-                             amount_cents=50, status="reviewing", source="coin_cash"))
-        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview",
-                             amount_cents=80, status="reviewing", source="invite_cash"))
         db.commit()
     finally:
         db.close()
@@ -388,106 +325,6 @@ def test_withdraw_list_exposes_source_and_filters(
     )
     items = r.json()["items"]
     assert items and all(it["source"] == "invite_cash" for it in items)
-    assert all(it["is_high_risk"] is True for it in items)
-    assert all(it["high_risk_note"] == "邀请行为异常" for it in items)
-    # 累计提现按当前审核页账户隔离，不把普通现金的 100 分串进邀请金。
-    assert all(it["cumulative_success_cents"] == 200 for it in items)
-
-    invite_summary = admin_client.get(
-        "/admin/api/withdraws/summary",
-        params={"source": "invite_cash"},
-        headers=_auth(admin_token),
-    ).json()
-    other_summary = admin_client.get(
-        "/admin/api/withdraws/summary",
-        params={"source": "coin_cash"},
-        headers=_auth(admin_token),
-    ).json()
-    assert invite_summary["reviewing_count"] >= 1
-    assert other_summary["reviewing_count"] >= 1
-    assert invite_summary["reviewing_amount_cents"] >= 80
-    assert other_summary["reviewing_amount_cents"] >= 50
-
-
-def test_invite_withdraw_detail_contains_invitee_first_actions(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    db = SessionLocal()
-    try:
-        inviter = user_repo.upsert_user_for_login(
-            db, phone="13800009031", register_channel="sms"
-        )
-        invitee = user_repo.upsert_user_for_login(
-            db, phone="13800009032", register_channel="sms"
-        )
-        db.add(
-            InviteRelation(
-                inviter_user_id=inviter.id,
-                invitee_user_id=invitee.id,
-                status="effective",
-                compare_reward_granted=True,
-                compare_reward_cents=50,
-            )
-        )
-        db.add(
-            ComparisonRecord(
-                user_id=invitee.id,
-                trace_id="invite-detail-first-compare",
-                status="success",
-                store_name="首次比价商家",
-                product_names="商品甲、商品乙",
-            )
-        )
-        db.add(
-            SavingsRecord(
-                user_id=invitee.id,
-                order_amount_cents=1888,
-                saved_amount_cents=300,
-                source="compare",
-                shop_name="首次下单商家",
-                title="首次下单商品",
-                dishes=[],
-            )
-        )
-        bill = "inviteoverviewbill0001"
-        db.add(
-            WithdrawOrder(
-                user_id=inviter.id,
-                out_bill_no=bill,
-                amount_cents=100,
-                status="reviewing",
-                source="invite_cash",
-            )
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    response = admin_client.get(
-        f"/admin/api/withdraws/{bill}", headers=_auth(admin_token)
-    )
-    assert response.status_code == 200, response.text
-    overview = response.json()["invite_overview"]
-    assert overview["invite_total"] == 1
-    assert overview["invite_success_total"] == 1
-    item = overview["items"][0]
-    assert item["phone"] == "13800009032"
-    assert item["first_compare_store"] == "首次比价商家"
-    assert item["first_compare_products"] == "商品甲、商品乙"
-    assert item["first_order_store"] == "首次下单商家"
-    assert item["first_order_products"] == "首次下单商品"
-    assert item["first_order_amount_cents"] == 1888
-
-    filtered = admin_client.get(
-        f"/admin/api/withdraws/{bill}",
-        params={
-            "date_from": "2099-01-01T00:00:00Z",
-            "date_to": "2099-01-02T00:00:00Z",
-        },
-        headers=_auth(admin_token),
-    )
-    assert filtered.status_code == 200, filtered.text
-    assert filtered.json()["invite_overview"]["invite_total"] == 0
 
 
 def test_ad_coin_audit_full_count_truncate_and_only_mismatch(

tests/test_admin_roles.py

@@ -68,17 +68,10 @@ 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"]] == [
-        "risk-monitor",
-        "device-liveness",
-        "analytics-health",
-        "event-logs",
-        "audit-logs",
+        "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
@@ -89,9 +82,8 @@ 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",
@@ -190,16 +182,10 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     assert roles["finance"]["label"] == "财务"
     assert roles["tech"]["label"] == "技术"
     # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
-    assert set(roles["finance"]["pages"]) == {
-        "dashboard",
-        "ad-revenue-report",
-        "cps",
-        "invite-withdraws",
-        "withdraws",
-    }
+    assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
     assert set(roles["tech"]["pages"]) == {
-        "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
-        "ad-revenue", "huawei-review", "event-logs", "audit-logs",
+        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
+        "event-logs", "audit-logs",
     }
 
 

tests/test_admin_write.py

@@ -103,61 +103,6 @@ def _seed_price_report(phone: str) -> int:
         db.close()
 
 
-def test_set_user_risk_requires_note_and_writes_audit(
-    admin_client: TestClient, operator_token: str
-) -> None:
-    uid = _seed_user("13900000991")
-
-    missing_note = admin_client.post(
-        f"/admin/api/users/{uid}/risk",
-        json={"is_high_risk": True, "note": "   "},
-        headers=_auth(operator_token),
-    )
-    assert missing_note.status_code == 422
-
-    marked = admin_client.post(
-        f"/admin/api/users/{uid}/risk",
-        json={"is_high_risk": True, "note": "邀请记录存在批量异常"},
-        headers=_auth(operator_token),
-    )
-    assert marked.status_code == 200, marked.text
-
-    listed = admin_client.get(
-        "/admin/api/users",
-        params={"phone": "13900000991"},
-        headers=_auth(operator_token),
-    ).json()["items"][0]
-    assert listed["is_high_risk"] is True
-    assert listed["high_risk_note"] == "邀请记录存在批量异常"
-
-    db = SessionLocal()
-    try:
-        audit = db.execute(
-            select(AdminAuditLog)
-            .where(
-                AdminAuditLog.action == "user.risk.set",
-                AdminAuditLog.target_id == str(uid),
-            )
-            .order_by(AdminAuditLog.id.desc())
-        ).scalars().first()
-        assert audit is not None
-        assert audit.detail["after"]["is_high_risk"] is True
-    finally:
-        db.close()
-
-    cleared = admin_client.post(
-        f"/admin/api/users/{uid}/risk",
-        json={"is_high_risk": False, "note": None},
-        headers=_auth(operator_token),
-    )
-    assert cleared.status_code == 200
-    overview = admin_client.get(
-        f"/admin/api/users/{uid}", headers=_auth(operator_token)
-    ).json()
-    assert overview["user"]["is_high_risk"] is False
-    assert overview["user"]["high_risk_note"] is None
-
-
 # ===== 调金币 =====
 
 def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
@@ -634,47 +579,15 @@ def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> N
     ).status_code == 404
 
 
-def test_withdraw_refresh_enriches_existing_failed_reason(
-    admin_client: TestClient, finance_token: str, monkeypatch
-) -> None:
-    uid = _seed_user("13900000017")
-    with SessionLocal() as db:
-        db.add(
-            WithdrawOrder(
-                user_id=uid,
-                out_bill_no="adminfailedreason01",
-                amount_cents=100,
-                status="failed",
-                wechat_state="FAIL",
-                fail_reason="微信转账状态 FAIL",
-            )
-        )
-        db.commit()
-
+def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
     from app.repositories import wallet as wr
-
     monkeypatch.setattr(
-        wr.wxpay,
-        "query_transfer",
-        lambda out_bill_no: {
-            "status_code": 200,
-            "data": {
-                "state": "FAIL",
-                "fail_reason": "TRANSFER_RISK",
-                "transfer_bill_no": "wx_failed_reason_01",
-            },
-        },
+        wr.wxpay, "query_transfer",
+        lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
     )
-
-    response = admin_client.post(
-        "/admin/api/withdraws/adminfailedreason01/refresh",
+    r = admin_client.post(
+        "/admin/api/withdraws/reconcile", params={"older_than_minutes": 0},
         headers=_auth(finance_token),
     )
-
-    assert response.status_code == 200, response.text
-    assert response.json()["status"] == "failed"
-    assert response.json()["wechat_state"] == "FAIL"
-    assert response.json()["transfer_bill_no"] == "wx_failed_reason_01"
-    assert response.json()["fail_reason"] == (
-        "微信转账失败：该笔转账存在风险，已被微信拦截（TRANSFER_RISK）"
-    )
+    assert r.status_code == 200, r.text
+    assert "checked" in r.json() and "resolved" in r.json()

tests/test_invite_cash_withdraw.py

@@ -127,51 +127,6 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None:
         db.close()
 
 
-def test_invite_cash_reject_recreates_missing_account(client, monkeypatch) -> None:
-    """A legacy or hand-written order without CoinAccount can still be refunded."""
-    _patch_userinfo(monkeypatch, "openid_ic_missing_account")
-    token = _login(client, "13800004012")
-    _seed_balances(client, token, "13800004012", cash=0, invite_cash=200)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    response = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    assert response.status_code == 200, response.text
-    bill = response.json()["out_bill_no"]
-
-    with SessionLocal() as db:
-        user = db.execute(
-            select(User).where(User.phone == "13800004012")
-        ).scalar_one()
-        db.delete(db.get(CoinAccount, user.id))
-        db.commit()
-
-    _reject(bill, "missing account fixture")
-
-    with SessionLocal() as db:
-        user = db.execute(
-            select(User).where(User.phone == "13800004012")
-        ).scalar_one()
-        account = db.get(CoinAccount, user.id)
-        order = crud_wallet._get_withdraw_or_raise(db, bill)
-        refunds = db.execute(
-            select(InviteCashTransaction).where(
-                InviteCashTransaction.user_id == user.id,
-                InviteCashTransaction.biz_type == "invite_withdraw_refund",
-                InviteCashTransaction.ref_id == bill,
-            )
-        ).scalars().all()
-
-        assert account is not None
-        assert account.invite_cash_balance_cents == 200
-        assert order.status == "rejected"
-        assert order.fail_reason == "missing account fixture"
-        assert len(refunds) == 1 and refunds[0].amount_cents == 200
-
-
 def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     """两账户各提各的不串:提 invite_cash(拒绝→验证退款回原账户),再提 cash,各扣各账户互不串。"""
     _patch_userinfo(monkeypatch, "openid_ic_3")

tests/test_risk_monitor.py

@@ -1,541 +0,0 @@
-"""风控监控：三类规则、明细及处置闭环。"""
-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,42 +190,6 @@ 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

@@ -9,7 +9,7 @@ from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.user import User
-from app.models.wallet import CoinAccount, WechatTransferAuthorization, WithdrawOrder
+from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
 from app.repositories import wallet as crud_wallet
 
 
@@ -232,78 +232,6 @@ def test_withdraw_approve_transfer_fail_refunds(client, monkeypatch) -> None:
     assert r.json()["items"][0]["status"] == "failed"
 
 
-def test_approved_withdraw_query_fail_exposes_wechat_reason(client, monkeypatch) -> None:
-    """审核通过后微信终态失败时，保存官方失败码对应原因并退回余额。"""
-    monkeypatch.setattr(
-        "app.integrations.wxpay.code_to_userinfo",
-        lambda code: {
-            "openid": "openid_query_fail",
-            "nickname": None,
-            "avatar_url": None,
-            "raw": {},
-        },
-    )
-    monkeypatch.setattr(
-        "app.integrations.wxpay.create_transfer",
-        lambda openid, amount_fen, out_bill_no, user_name=None: {
-            "status_code": 200,
-            "data": {
-                "state": "WAIT_USER_CONFIRM",
-                "package_info": "pkg_query_fail",
-                "transfer_bill_no": "tb_query_fail",
-            },
-        },
-    )
-    token = _login(client, "13800002025")
-    _seed_cash(client, token, "13800002025", 100)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    response = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 50},
-        headers=_auth(token),
-    )
-    bill = response.json()["out_bill_no"]
-    _approve(bill)
-
-    monkeypatch.setattr(
-        "app.integrations.wxpay.query_transfer",
-        lambda out_bill_no: {
-            "status_code": 200,
-            "data": {
-                "state": "FAIL",
-                "fail_reason": "PAYEE_ACCOUNT_ABNORMAL",
-                "transfer_bill_no": "tb_query_fail",
-            },
-        },
-    )
-    response = client.get(
-        "/api/v1/wallet/withdraw/status",
-        params={"out_bill_no": bill},
-        headers=_auth(token),
-    )
-
-    assert response.status_code == 200, response.text
-    assert response.json()["status"] == "failed"
-    assert response.json()["wechat_state"] == "FAIL"
-    assert response.json()["fail_reason"] == (
-        "微信转账失败：用户微信账户收款异常（PAYEE_ACCOUNT_ABNORMAL）"
-    )
-    with SessionLocal() as db:
-        order = db.execute(
-            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill)
-        ).scalar_one()
-        assert order.transfer_bill_no == "tb_query_fail"
-    account = client.get("/api/v1/wallet/account", headers=_auth(token))
-    assert account.json()["cash_balance_cents"] == 100
-
-
-def test_unknown_wechat_fail_reason_code_is_preserved() -> None:
-    assert crud_wallet._wechat_terminal_failure_reason(
-        {"fail_reason": "NEW_WECHAT_REASON"}, "FAIL"
-    ) == "微信转账失败：NEW_WECHAT_REASON"
-
-
 def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
     """#2 同 out_bill_no 重试:只转一次,第二次返回同一单,余额只扣一次。"""
     monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_idem", "nickname": None, "avatar_url": None, "raw": {}})

tests/test_withdraw_ledger_check.py

@@ -0,0 +1,151 @@
+"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。
+
+历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对,
+而 source=invite_cash 的提现单流水其实在 invite_cash_transaction 表,导致每笔邀请提现单都被
+误报「缺扣款/缺退款流水」。这里用真实提现 API 造单 + before/after 差值断言锁定修复:
+  1) 邀请提现单不再污染普通现金账的缺流水计数;
+  2) 邀请账户已被纳入对账(能抓到它自己的缺流水);
+  3) 普通现金账的原有对账未被改坏。
+
+conftest 的库是 session 级共享、测试间不清,故一律用 before/after 差值,只反映本用例造的数据。
+"""
+from __future__ import annotations
+
+from sqlalchemy import delete, select
+
+from app.admin.repositories.queries import withdraw_ledger_check
+from app.db.session import SessionLocal
+from app.models.user import User
+from app.models.wallet import CoinAccount, InviteCashTransaction
+from app.repositories import wallet as crud_wallet
+
+
+def _login(client, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _patch_userinfo(monkeypatch, openid: str) -> None:
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}},
+    )
+
+
+def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None:
+    client.get("/api/v1/wallet/account", headers=_auth(token))
+    db = SessionLocal()
+    try:
+        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
+        acc = db.get(CoinAccount, user.id)
+        acc.cash_balance_cents = cash
+        acc.invite_cash_balance_cents = invite_cash
+        db.commit()
+    finally:
+        db.close()
+
+
+def _reject(bill: str, reason: str = "测试拒绝") -> None:
+    db = SessionLocal()
+    try:
+        crud_wallet.reject_withdraw(db, bill, reason)
+    finally:
+        db.close()
+
+
+def _ledger() -> dict:
+    db = SessionLocal()
+    try:
+        return withdraw_ledger_check(db)
+    finally:
+        db.close()
+
+
+def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None:
+    """核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction,
+    不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)。"""
+    before = _ledger()
+
+    _patch_userinfo(monkeypatch, "openid_lc_1")
+    token = _login(client, "13800005001")
+    _seed_balances(client, token, "13800005001", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # rejected + 退款流水落 invite_cash_transaction
+
+    after = _ledger()
+
+    # 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报)
+    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
+    assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"]
+    # 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺
+    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]
+    assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"]
+
+
+def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None:
+    """删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False,
+    证明邀请账户已真正纳入对账(修复前邀请账完全不校验、永远报不出问题)。"""
+    _patch_userinfo(monkeypatch, "openid_lc_2")
+    token = _login(client, "13800005002")
+    _seed_balances(client, token, "13800005002", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    bill = r.json()["out_bill_no"]
+
+    before = _ledger()
+    db = SessionLocal()
+    try:
+        db.execute(
+            delete(InviteCashTransaction).where(
+                InviteCashTransaction.ref_id == bill,
+                InviteCashTransaction.biz_type == "invite_withdraw",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+    after = _ledger()
+
+    assert (
+        after["invite_missing_withdraw_txn_count"]
+        == before["invite_missing_withdraw_txn_count"] + 1
+    )
+    assert after["ok"] is False
+
+
+def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
+    """普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。"""
+    _patch_userinfo(monkeypatch, "openid_lc_3")
+    token = _login(client, "13800005003")
+    _seed_balances(client, token, "13800005003", cash=500, invite_cash=0)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    before = _ledger()
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    after = _ledger()
+
+    # 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响
+    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
+    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]