修复：领券单券成功率按任务独立统计

## 背景

PR #162 已合并。本 PR 为其后续日志增强，方便人工按一次任务完整回查美团、京东每日自动对账。

## 日志内容

- 每次运行生成唯一 `run_id`，记录 `scheduled` / `startup_catchup` 触发来源
- 记录北京时间计划日期、近 3 天回拉窗口、环境、数据库方言、主机名和 PID
- 分平台记录开始、成功、跳过、失败及执行耗时
- 成功结果记录 fetched / inserted / updated / pages / api_requests，京东额外记录小时窗口数
- 京东上游失败记录具体小时窗口、页码和请求序号；美团记录失败页码
- 最终汇总记录 success / partial_success / failed、失败平台、是否需要人工补跑和下次执行时间
- 日志使用现有 `extra` 结构化字段写入 JSON 日志，便于 SLS/人工检索

## 兼容性

- 手动对账接口、事务和返回模型不变
- 不记录密钥、Token、完整上游响应或订单明细
- 仓储层仅增加可选审计上下文和请求计数；不改变拉取及 upsert 逻辑

## 验证

- `pytest tests/test_cps_reconcile_worker.py tests/test_cps_admin.py tests/test_admin_read.py tests/test_observe.py -q`：44 passed
- 相关文件 Ruff 检查通过（忽略文件原有 UP017 提示）
- `git diff --check` 通过

---------

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

.env.example

@@ -113,6 +113,13 @@ JD_UNION_APP_SECRET=
 JD_UNION_SITE_ID=
 JD_UNION_AUTH_KEY=
 
+# 美团 + 京东订单每天北京时间 05:00 自动对账；按更新时间回拉近 3 天，重叠防漏单并刷新状态。
+# 手动对账按钮不受该开关影响。通常保持开启；临时停自动任务时设为 false。
+CPS_AUTO_RECONCILE_ENABLED=true
+CPS_AUTO_RECONCILE_RUN_HOUR=5
+CPS_AUTO_RECONCILE_LOOKBACK_DAYS=3
+CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC=60
+
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。

alembic/versions/8e04cc13a211_merge_notification_comparison_user_idx_.py

@@ -0,0 +1,26 @@
+"""merge notification/comparison_user_idx/monitoring_audit heads
+
+Revision ID: 8e04cc13a211
+Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table
+Create Date: 2026-07-23 15:37:26.967540
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '8e04cc13a211'
+down_revision: Union[str, Sequence[str], None] = ('comparison_user_created_idx', 'monitoring_audit_rbac', 'notification_table')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

alembic/versions/comparison_user_created_idx.py

@@ -0,0 +1,52 @@
+"""add composite index (user_id, created_at, id) on comparison_record
+
+C 端「我的比价记录」列表(GET /api/v1/compare/records)是
+`WHERE user_id=? ORDER BY created_at DESC, id DESC LIMIT n` —— 原来只有单列 user_id 索引,
+过滤完还要把该用户的**全部**记录取出来排序才能拿前 n 条,重度用户随记录数线性变慢。
+
+本复合索引的反向扫恰好等于 (created_at DESC, id DESC),规划器直接取前 n 条、免排序。
+列序 (user_id, created_at, id) 与查询一一对应,不要调整。
+
+Revision ID: comparison_user_created_idx
+Revises: merge_active_phone
+Create Date: 2026-07-21
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "comparison_user_created_idx"
+down_revision = "merge_active_phone"
+branch_labels = None
+depends_on = None
+
+INDEX_NAME = "ix_comparison_user_created"
+COLUMNS = ["user_id", "created_at", "id"]
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name == "postgresql":
+        # 线上 comparison_record 已有数据量,普通 CREATE INDEX 持表写锁会阻塞比价 harvest 写入;
+        # 用 CONCURRENTLY 不锁表(须脱离事务,autocommit_block 切到自动提交)。
+        # 同 comparison_status_created_idx 的做法。
+        with op.get_context().autocommit_block():
+            op.create_index(
+                INDEX_NAME, "comparison_record", COLUMNS,
+                unique=False, postgresql_concurrently=True,
+            )
+    else:
+        op.create_index(INDEX_NAME, "comparison_record", COLUMNS, unique=False)
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name == "postgresql":
+        with op.get_context().autocommit_block():
+            op.drop_index(
+                INDEX_NAME, table_name="comparison_record",
+                postgresql_concurrently=True,
+            )
+    else:
+        op.drop_index(INDEX_NAME, table_name="comparison_record")

alembic/versions/coupon_claim_event.py

@@ -0,0 +1,93 @@
+"""add per-session coupon claim event table
+
+Revision ID: coupon_claim_event
+Revises: 8e04cc13a211
+Create Date: 2026-07-23
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "coupon_claim_event"
+down_revision: str | Sequence[str] | None = "8e04cc13a211"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "coupon_claim_event",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("trace_id", sa.String(length=64), nullable=False),
+        sa.Column("device_id", sa.String(length=64), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("coupon_id", sa.String(length=64), nullable=False),
+        sa.Column("claim_date", sa.Date(), nullable=False),
+        sa.Column("status", sa.String(length=24), nullable=False),
+        sa.Column("app_env", sa.String(length=16), nullable=True),
+        sa.Column("vendor", sa.String(length=48), nullable=True),
+        sa.Column("coupon_name", sa.String(length=128), nullable=True),
+        sa.Column("claimed_count", sa.Integer(), nullable=True),
+        sa.Column("reason", sa.String(length=255), nullable=True),
+        sa.Column("extra", _JSON, nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "trace_id", "coupon_id",
+            name="uq_coupon_claim_event_trace_coupon",
+        ),
+    )
+    op.create_index(
+        "ix_coupon_claim_event_date_env",
+        "coupon_claim_event",
+        ["claim_date", "app_env"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_coupon_claim_event_app_env"),
+        "coupon_claim_event",
+        ["app_env"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_coupon_claim_event_trace_id"),
+        "coupon_claim_event",
+        ["trace_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_coupon_claim_event_user_id"),
+        "coupon_claim_event",
+        ["user_id"],
+        unique=False,
+    )
+
+    # 旧表只能回填当前仍保留的 trace；历史上已被每日去重覆盖的关联无法恢复。
+    op.execute(
+        """
+        INSERT INTO coupon_claim_event (
+            trace_id, device_id, user_id, coupon_id, claim_date, status, app_env,
+            vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
+        )
+        SELECT
+            trace_id, device_id, user_id, coupon_id, claim_date, status, app_env,
+            vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
+        FROM coupon_claim_record
+        WHERE trace_id IS NOT NULL
+        """
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_coupon_claim_event_user_id"), table_name="coupon_claim_event")
+    op.drop_index(op.f("ix_coupon_claim_event_trace_id"), table_name="coupon_claim_event")
+    op.drop_index(op.f("ix_coupon_claim_event_app_env"), table_name="coupon_claim_event")
+    op.drop_index("ix_coupon_claim_event_date_env", table_name="coupon_claim_event")
+    op.drop_table("coupon_claim_event")

alembic/versions/monitoring_audit_rbac.py

@@ -0,0 +1,56 @@
+"""补齐监控审计页面权限。
+
+Revision ID: monitoring_audit_rbac
+Revises: merge_signin_boost_main
+Create Date: 2026-07-22 00:00:00.000000
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "monitoring_audit_rbac"
+down_revision: str | Sequence[str] | None = "merge_signin_boost_main"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+_PAGE = "analytics-health"
+
+
+def _role_table() -> sa.TableClause:
+    return sa.table(
+        "admin_role",
+        sa.column("name", sa.String),
+        sa.column("pages", _JSON),
+    )
+
+
+def upgrade() -> None:
+    role = _role_table()
+    conn = op.get_bind()
+    pages = conn.execute(
+        sa.select(role.c.pages).where(role.c.name == "tech")
+    ).scalar_one_or_none()
+    if pages is not None and _PAGE not in pages:
+        conn.execute(
+            role.update()
+            .where(role.c.name == "tech")
+            .values(pages=[*pages, _PAGE])
+        )
+
+
+def downgrade() -> None:
+    role = _role_table()
+    conn = op.get_bind()
+    pages = conn.execute(
+        sa.select(role.c.pages).where(role.c.name == "tech")
+    ).scalar_one_or_none()
+    if pages is not None and _PAGE in pages:
+        conn.execute(
+            role.update()
+            .where(role.c.name == "tech")
+            .values(pages=[page for page in pages if page != _PAGE])
+        )

app/admin/deps.py

@@ -10,6 +10,8 @@ from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
 
+from app.admin.permissions import ALL_PAGE_KEYS, CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages
+from app.admin.repositories import admin_role as role_repo
 from app.admin.repositories import admin_user as admin_repo
 from app.admin.security import AdminTokenError, decode_admin_token
 from app.db.session import get_db
@@ -72,6 +74,33 @@ def require_role(*roles: str):
     return _checker
 
 
+def require_page(page: str):
+    """页面权限守卫依赖工厂。
+
+    左侧导航隐藏只是 UI，这个守卫确保直接调用 API 也必须持有对应页面权限。
+    super_admin 恒通过；custom 读个人 pages_override；其余角色读 admin_role.pages。
+    """
+    if page not in ALL_PAGE_KEYS:
+        raise ValueError(f"unknown admin page permission: {page}")
+
+    def _checker(admin: CurrentAdmin, db: AdminDb) -> AdminUser:
+        if admin.role == SUPER_ADMIN_ROLE:
+            return admin
+        pages = (
+            sanitize_pages(admin.pages_override)
+            if admin.role == CUSTOM_ROLE
+            else role_repo.effective_pages_of(db, admin.role)
+        )
+        if page not in pages:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"page '{page}' not allowed",
+            )
+        return admin
+
+    return _checker
+
+
 def get_client_ip(request: Request) -> str:
     """取客户端 IP(审计日志用)。生产经 nginx 反代,优先 X-Forwarded-For 第一段;否则直连 IP。
 

app/admin/permissions.py

@@ -21,7 +21,6 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "ad-revenue-report", "label": "广告收益"},
         {"key": "comparison-records", "label": "比价记录"},
         {"key": "cps", "label": "CPS收益"},
-        {"key": "device-liveness", "label": "设备存活"},
     ]},
     {"group": "奖励审核", "pages": [
         {"key": "withdraws", "label": "提现审核"},
@@ -34,11 +33,15 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "huawei-review", "label": "华为审核开关"},
         {"key": "users", "label": "用户管理"},
     ]},
-    {"group": "其他", "pages": [
-        {"key": "admins", "label": "权限管理"},
+    {"group": "监控审计", "pages": [
+        {"key": "device-liveness", "label": "设备存活"},
+        {"key": "analytics-health", "label": "埋点成功率"},
         {"key": "event-logs", "label": "埋点日志"},
         {"key": "audit-logs", "label": "审计日志"},
     ]},
+    {"group": "其他", "pages": [
+        {"key": "admins", "label": "权限管理"},
+    ]},
 ]
 
 # 全部页面 key(super_admin 有效可见 = 此全集;也用于校验角色 pages 合法性)
@@ -58,7 +61,7 @@ BUILTIN_ROLES: list[dict] = [
         "dashboard", "ad-revenue-report", "cps", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
-        "dashboard", "device-liveness", "config", "ad-revenue", "huawei-review",
+        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
         "event-logs", "audit-logs",
     ]},
 ]

app/admin/repositories/ad_revenue.py

@@ -81,6 +81,10 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 # ad_feed_reward_record,由 audit 内部按 ad_type 区分(feed 含历史 NULL,draw 仅 ad_type=="draw")。
 _AUDIT_SCENES = {"reward_video", "feed", "draw"}
 
+# 激励视频未满足有效播放条件时不计客户端预估收益。客户端仍会在 onAdShow
+# 上报 eCPM，随后才在关闭时补报以下终态，因此必须在展示/发奖合并后修正收益。
+_ZERO_REVENUE_REWARD_VIDEO_STATUSES = frozenset({"closed_early", "too_short"})
+
 
 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
 _REWARD_DETAIL_KEYS = (
@@ -202,6 +206,11 @@ def ad_revenue_report(
                 "matched": bool(rwd["matched"]),
                 "reward_detail": _reward_detail(rwd),
             })
+            if (
+                rec.ad_type == "reward_video"
+                and rwd["status"] in _ZERO_REVENUE_REWARD_VIDEO_STATUSES
+            ):
+                ev["revenue_yuan"] = 0.0
         else:
             # 纯展示(信息流逐条展示、激励视频缺发奖记录):不计对账,matched=True。
             ev.update({

app/admin/repositories/coupon_data.py

@@ -1,7 +1,7 @@
 """admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。
 
-数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
-区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
+数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。生产 PostgreSQL
+使用 percentile_cont 聚合耗时分位；SQLite 本地/测试环境回退读取耗时单列计算。
 - 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
 - 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
 - summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
@@ -16,7 +16,7 @@ from sqlalchemy import case, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
-from app.models.coupon_state import CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
 from app.models.user import User
 from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
@@ -45,6 +45,75 @@ def _percentile(sorted_vals: list[int], q: float) -> int | None:
     return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)
 
 
+def _round_duration_ms(value) -> int | None:
+    """将数据库聚合结果按既有 Python round 口径转为整数毫秒。"""
+    if value is None:
+        return None
+    return int(round(value))
+
+
+def _coupon_summary_aggregate_stmt(conditions: list):
+    """PostgreSQL 汇总卡聚合语句；计数、均值与四个分位一次返回。"""
+    completed = CouponSession.status == "completed"
+    completed_elapsed = completed & CouponSession.elapsed_ms.is_not(None)
+    return select(
+        func.count(CouponSession.id),
+        func.sum(case((completed, 1), else_=0)),
+        func.avg(CouponSession.elapsed_ms).filter(completed_elapsed),
+        *(
+            func.percentile_cont(q)
+            .within_group(CouponSession.elapsed_ms)
+            .filter(completed_elapsed)
+            for q in (0.05, 0.5, 0.95, 0.99)
+        ),
+    ).where(*conditions)
+
+
+def _coupon_summary_aggregates(db: Session, conditions: list) -> dict:
+    """汇总卡基础指标；生产 PG 全部在数据库内完成，SQLite 仅作测试回退。"""
+    if db.bind is not None and db.bind.dialect.name == "postgresql":
+        row = db.execute(_coupon_summary_aggregate_stmt(conditions)).one()
+        return {
+            "started_count": int(row[0] or 0),
+            "completed_count": int(row[1] or 0),
+            "avg_elapsed_ms": _round_duration_ms(row[2]),
+            "p5_ms": _round_duration_ms(row[3]),
+            "p50_ms": _round_duration_ms(row[4]),
+            "p95_ms": _round_duration_ms(row[5]),
+            "p99_ms": _round_duration_ms(row[6]),
+        }
+
+    counts = db.execute(
+        select(
+            func.count(CouponSession.id),
+            func.sum(case((CouponSession.status == "completed", 1), else_=0)),
+        ).where(*conditions)
+    ).one()
+    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
+    completed_elapsed = list(
+        db.execute(
+            select(CouponSession.elapsed_ms)
+            .where(
+                *conditions,
+                CouponSession.status == "completed",
+                CouponSession.elapsed_ms.is_not(None),
+            )
+            .order_by(CouponSession.elapsed_ms)
+        ).scalars()
+    )
+    return {
+        "started_count": int(counts[0] or 0),
+        "completed_count": int(counts[1] or 0),
+        "avg_elapsed_ms": _round_duration_ms(
+            sum(completed_elapsed) / len(completed_elapsed)
+        ) if completed_elapsed else None,
+        "p5_ms": _percentile(completed_elapsed, 5),
+        "p50_ms": _percentile(completed_elapsed, 50),
+        "p95_ms": _percentile(completed_elapsed, 95),
+        "p99_ms": _percentile(completed_elapsed, 99),
+    }
+
+
 def _avg(vals: list[int]) -> int | None:
     return round(sum(vals) / len(vals)) if vals else None
 
@@ -124,18 +193,18 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
     """聚合查询批量返回逐场点位分数，不加载逐券明细。"""
     if not trace_ids:
         return {}
-    succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
+    succeeded = func.sum(case((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
     rows = db.execute(
         select(
-            CouponClaimRecord.trace_id,
+            CouponClaimEvent.trace_id,
             succeeded.label("succeeded"),
             func.count().label("tried"),
         )
         .where(
-            CouponClaimRecord.trace_id.in_(trace_ids),
-            CouponClaimRecord.status.in_(_SLOT_TRIED),
+            CouponClaimEvent.trace_id.in_(trace_ids),
+            CouponClaimEvent.status.in_(_SLOT_TRIED),
         )
-        .group_by(CouponClaimRecord.trace_id)
+        .group_by(CouponClaimEvent.trace_id)
     ).all()
     return {
         trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
@@ -148,13 +217,13 @@ def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
     """按单个 trace 查询逐券结果；仅在后台用户点击分数时调用。"""
     rows = db.execute(
         select(
-            CouponClaimRecord.coupon_id,
-            CouponClaimRecord.coupon_name,
-            CouponClaimRecord.status,
-            CouponClaimRecord.reason,
+            CouponClaimEvent.coupon_id,
+            CouponClaimEvent.coupon_name,
+            CouponClaimEvent.status,
+            CouponClaimEvent.reason,
         )
-        .where(CouponClaimRecord.trace_id == trace_id)
-        .order_by(CouponClaimRecord.id)
+        .where(CouponClaimEvent.trace_id == trace_id)
+        .order_by(CouponClaimEvent.id)
     ).all()
     return [
         {
@@ -215,30 +284,22 @@ def coupon_data_report(
         if not user_ids:
             return _empty_result()
 
-    stmt = select(CouponSession).where(
+    conditions = [
         CouponSession.started_date >= d_from,
         CouponSession.started_date <= d_to,
-    )
+    ]
     if app_env is not None:
-        stmt = stmt.where(CouponSession.app_env == app_env)
+        conditions.append(CouponSession.app_env == app_env)
     if statuses:
-        stmt = stmt.where(CouponSession.status.in_(statuses))
+        conditions.append(CouponSession.status.in_(statuses))
     if user_ids is not None:
-        stmt = stmt.where(CouponSession.user_id.in_(user_ids))
+        conditions.append(CouponSession.user_id.in_(user_ids))
+    stmt = select(CouponSession).where(*conditions)
     rows = list(db.execute(stmt).scalars())
 
     # ── 汇总卡 ──
-    completed_elapsed = sorted(
-        r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None
-    )
     summary = {
-        "started_count": len(rows),
-        "completed_count": sum(1 for r in rows if r.status == "completed"),
-        "avg_elapsed_ms": _avg(completed_elapsed),
-        "p5_ms": _percentile(completed_elapsed, 5),
-        "p50_ms": _percentile(completed_elapsed, 50),
-        "p95_ms": _percentile(completed_elapsed, 95),
-        "p99_ms": _percentile(completed_elapsed, 99),
+        **_coupon_summary_aggregates(db, conditions),
         **_success_rates(rows),
     }
 
@@ -323,7 +384,7 @@ def coupon_data_report(
         "summary": summary,
         "daily": daily,
         "hourly": hourly,
-        "total": len(rows),
+        "total": summary["started_count"],
         "items": items,
     }
 

app/admin/repositories/cps.py

@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from typing import Any
@@ -15,12 +16,14 @@ from sqlalchemy.orm import Session
 
 from app.admin.repositories.queries import _as_utc, offset_paginate
 from app.integrations import jd_union, meituan
-from app.repositories import cps_link as cps_link_repo
 from app.models.cps_activity import CpsActivity
 from app.models.cps_group import CpsGroup
 from app.models.cps_link import CpsClick, CpsLink
 from app.models.cps_order import CpsOrder
 from app.models.cps_wx_user import CpsWxUser
+from app.repositories import cps_link as cps_link_repo
+
+logger = logging.getLogger("shagua.cps_reconcile")
 
 # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
 _INVALID_STATUS = {"4", "5"}
@@ -378,18 +381,35 @@ def effective_commission_cents(order: CpsOrder) -> int:
 def reconcile_orders(
     db: Session, *, start_time: int, end_time: int,
     query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
+    audit_context: dict[str, Any] | None = None,
 ) -> dict:
     """调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
 
     订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
     """
-    fetched = inserted = updated = pages = 0
+    fetched = inserted = updated = pages = api_requests = 0
     page = 1
     while page <= max_pages:
-        resp = meituan.query_order(
-            sid=sid, start_time=start_time, end_time=end_time,
-            query_time_type=query_time_type, page=page, limit=100,
-        )
+        api_requests += 1
+        try:
+            resp = meituan.query_order(
+                sid=sid, start_time=start_time, end_time=end_time,
+                query_time_type=query_time_type, page=page, limit=100,
+            )
+        except Exception:  # noqa: BLE001 - 记录失败页后保持原异常类型继续抛出
+            if audit_context is not None:
+                logger.exception(
+                    "CPS reconcile upstream request failed platform=meituan page=%s",
+                    page,
+                    extra={
+                        **audit_context,
+                        "event": "cps_reconcile.request_failed",
+                        "platform": "meituan",
+                        "failed_page": page,
+                        "api_request_number": api_requests,
+                    },
+                )
+            raise
         rows = ((resp.get("data") or {}).get("dataList")) or []
         if not rows:
             break
@@ -419,30 +439,58 @@ def reconcile_orders(
             break
         page += 1
     db.commit()
-    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
+    return {
+        "fetched": fetched,
+        "inserted": inserted,
+        "updated": updated,
+        "pages": pages,
+        "api_requests": api_requests,
+    }
 
 
 def reconcile_jd_orders(
     db: Session, *, start_time: datetime, end_time: datetime,
     query_time_type: int = 3, max_pages: int = 100,
+    audit_context: dict[str, Any] | None = None,
 ) -> dict:
     """调京东 order.row.query 拉单 → 按订单行 upsert。
 
     京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
     """
-    fetched = inserted = updated = pages = 0
+    fetched = inserted = updated = pages = api_requests = windows = 0
     cur = start_time
     while cur < end_time:
         win_end = min(cur + timedelta(hours=1), end_time)
+        windows += 1
         page = 1
         while page <= max_pages:
-            resp = jd_union.query_order_rows(
-                start_time=cur,
-                end_time=win_end,
-                query_time_type=query_time_type,
-                page_index=page,
-                page_size=200,
-            )
+            api_requests += 1
+            try:
+                resp = jd_union.query_order_rows(
+                    start_time=cur,
+                    end_time=win_end,
+                    query_time_type=query_time_type,
+                    page_index=page,
+                    page_size=200,
+                )
+            except Exception:  # noqa: BLE001 - 记录失败窗口后保持原异常类型继续抛出
+                if audit_context is not None:
+                    logger.exception(
+                        "CPS reconcile upstream request failed platform=jd window=%s..%s page=%s",
+                        cur.isoformat(),
+                        win_end.isoformat(),
+                        page,
+                        extra={
+                            **audit_context,
+                            "event": "cps_reconcile.request_failed",
+                            "platform": "jd",
+                            "failed_window_start": cur.isoformat(),
+                            "failed_window_end": win_end.isoformat(),
+                            "failed_page": page,
+                            "api_request_number": api_requests,
+                        },
+                    )
+                raise
             rows = resp.get("rows") or []
             has_more = bool(resp.get("has_more"))
             if not rows:
@@ -469,7 +517,14 @@ def reconcile_jd_orders(
             page += 1
         cur = win_end
     db.commit()
-    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
+    return {
+        "fetched": fetched,
+        "inserted": inserted,
+        "updated": updated,
+        "pages": pages,
+        "api_requests": api_requests,
+        "windows": windows,
+    }
 
 
 def list_orders(

app/admin/repositories/queries.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from datetime import date, datetime, time, timedelta, timezone
+from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import Select, asc, case, desc, func, or_, select
@@ -297,6 +298,58 @@ def _comparison_percentile(sorted_values: list[int], q: float) -> int | None:
     return int(value + 0.5)
 
 
+def _round_duration_ms(value) -> int | None:
+    """将数据库聚合结果按既有口径四舍五入为整数毫秒。"""
+    if value is None:
+        return None
+    return int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
+
+
+def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles: tuple[float, ...]):
+    """PostgreSQL 耗时聚合语句；每种状态只返回一行。"""
+    return select(
+        func.avg(ComparisonRecord.total_ms),
+        *(
+            func.percentile_cont(q).within_group(ComparisonRecord.total_ms)
+            for q in quantiles
+        ),
+    ).where(
+        *conditions,
+        ComparisonRecord.status == status,
+        ComparisonRecord.total_ms.is_not(None),
+    )
+
+
+def _comparison_duration_aggregates(
+    db: Session,
+    *,
+    conditions: list,
+    status: str,
+    quantiles: tuple[float, ...],
+) -> list[int | None]:
+    """返回平均值和各分位数；生产 PG 在数据库内聚合，SQLite 仅作测试回退。"""
+    if db.bind is not None and db.bind.dialect.name == "postgresql":
+        row = db.execute(
+            _comparison_duration_aggregate_stmt(conditions, status, quantiles)
+        ).one()
+        return [_round_duration_ms(value) for value in row]
+
+    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
+    values = list(
+        db.execute(
+            select(ComparisonRecord.total_ms)
+            .where(
+                *conditions,
+                ComparisonRecord.status == status,
+                ComparisonRecord.total_ms.is_not(None),
+            )
+            .order_by(ComparisonRecord.total_ms)
+        ).scalars()
+    )
+    average = _round_duration_ms(sum(values) / len(values)) if values else None
+    return [average, *(_comparison_percentile(values, q) for q in quantiles)]
+
+
 def comparison_records_summary(
     db: Session,
     *,
@@ -332,20 +385,18 @@ def comparison_records_summary(
     success = int(row[2] or 0)
     lower_price = int(row[4] or 0)
     cancelled = int(row[5] or 0)
-    success_durations = sorted(db.execute(
-        select(ComparisonRecord.total_ms).where(
-            *conditions,
-            ComparisonRecord.status == "success",
-            ComparisonRecord.total_ms.is_not(None),
-        )
-    ).scalars().all())
-    cancelled_durations = sorted(db.execute(
-        select(ComparisonRecord.total_ms).where(
-            *conditions,
-            ComparisonRecord.status == "cancelled",
-            ComparisonRecord.total_ms.is_not(None),
-        )
-    ).scalars().all())
+    success_duration_stats = _comparison_duration_aggregates(
+        db,
+        conditions=conditions,
+        status="success",
+        quantiles=(0.05, 0.5, 0.95, 0.99),
+    )
+    cancelled_duration_stats = _comparison_duration_aggregates(
+        db,
+        conditions=conditions,
+        status="cancelled",
+        quantiles=(0.05, 0.5, 0.95),
+    )
     success_rate_denominator = started - cancelled
     return {
         "started": started,
@@ -354,19 +405,16 @@ def comparison_records_summary(
         "success_rate": success / success_rate_denominator if success_rate_denominator else None,
         "avg_token_cost": float(row[3]) if row[3] is not None else None,
         "lower_price_rate": lower_price / success if success else None,
-        "avg_duration_ms": (
-            int(sum(success_durations) / len(success_durations) + 0.5)
-            if success_durations else None
-        ),
-        "p5_duration_ms": _comparison_percentile(success_durations, 0.05),
-        "p50_duration_ms": _comparison_percentile(success_durations, 0.5),
-        "p95_duration_ms": _comparison_percentile(success_durations, 0.95),
-        "p99_duration_ms": _comparison_percentile(success_durations, 0.99),
+        "avg_duration_ms": success_duration_stats[0],
+        "p5_duration_ms": success_duration_stats[1],
+        "p50_duration_ms": success_duration_stats[2],
+        "p95_duration_ms": success_duration_stats[3],
+        "p99_duration_ms": success_duration_stats[4],
         "cancelled": cancelled,
         "cancelled_rate": cancelled / started if started else None,
-        "cancelled_p5_ms": _comparison_percentile(cancelled_durations, 0.05),
-        "cancelled_p50_ms": _comparison_percentile(cancelled_durations, 0.5),
-        "cancelled_p95_ms": _comparison_percentile(cancelled_durations, 0.95),
+        "cancelled_p5_ms": cancelled_duration_stats[1],
+        "cancelled_p50_ms": cancelled_duration_stats[2],
+        "cancelled_p95_ms": cancelled_duration_stats[3],
     }
 
 

app/admin/routers/analytics_health.py

@@ -6,7 +6,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, get_current_admin
+from app.admin.deps import AdminDb, require_page
 from app.admin.repositories import analytics_health as repo
 from app.admin.schemas.analytics_health import (
     HealthBreakdownRow,
@@ -17,7 +17,7 @@ from app.admin.schemas.analytics_health import (
 router = APIRouter(
     prefix="/admin/api/analytics-health",
     tags=["admin-analytics-health"],
-    dependencies=[Depends(get_current_admin)],
+    dependencies=[Depends(require_page("analytics-health"))],
 )
 
 

app/admin/routers/audit.py

@@ -1,11 +1,11 @@
-"""admin 操作审计日志查询(所有 admin 可看:谁在何时对什么做了什么)。"""
+"""admin 操作审计日志查询(需要 audit-logs 页面权限)。"""
 from __future__ import annotations
 
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, get_current_admin
+from app.admin.deps import AdminDb, require_page
 from app.admin.repositories import audit_log as audit_repo
 from app.admin.schemas.admin import AdminAuditLogOut
 from app.admin.schemas.common import CursorPage
@@ -13,7 +13,7 @@ from app.admin.schemas.common import CursorPage
 router = APIRouter(
     prefix="/admin/api/audit-logs",
     tags=["admin-audit"],
-    dependencies=[Depends(get_current_admin)],
+    dependencies=[Depends(require_page("audit-logs"))],
 )
 
 

app/admin/routers/device_liveness.py

@@ -2,7 +2,7 @@
 
 数据源 device_liveness 表(心跳 last_heartbeat_at + liveness_state + kill_alert_pending,
 见 app/models/device.py)。在线/掉线、掉线时长由 repo 按 HEARTBEAT_TIMEOUT_MINUTES 阈值派生。
-纯读:无写、无审计。任意登录管理员可看(同大盘/设备管理,无角色门)。
+纯读:无写、无审计。需要 device-liveness 页面权限。
 """
 from __future__ import annotations
 
@@ -10,7 +10,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, get_current_admin
+from app.admin.deps import AdminDb, require_page
 from app.admin.repositories import queries
 from app.admin.schemas.common import CursorPage
 from app.admin.schemas.device import DeviceLivenessItem, DeviceLivenessStats
@@ -18,7 +18,7 @@ from app.admin.schemas.device import DeviceLivenessItem, DeviceLivenessStats
 router = APIRouter(
     prefix="/admin/api/device-liveness",
     tags=["admin-device-liveness"],
-    dependencies=[Depends(get_current_admin)],
+    dependencies=[Depends(require_page("device-liveness"))],
 )
 
 

app/admin/routers/event_logs.py

@@ -6,7 +6,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from app.admin.deps import AdminDb, get_current_admin
+from app.admin.deps import AdminDb, require_page
 from app.admin.repositories import queries
 from app.admin.schemas.analytics import AnalyticsEventOut
 from app.admin.schemas.common import CursorPage
@@ -14,7 +14,7 @@ from app.admin.schemas.common import CursorPage
 router = APIRouter(
     prefix="/admin/api/event-logs",
     tags=["admin-event-logs"],
-    dependencies=[Depends(get_current_admin)],
+    dependencies=[Depends(require_page("event-logs"))],
 )
 
 

app/admin/schemas/ad_revenue.py

@@ -26,7 +26,10 @@ class AdRevenueRecord(BaseModel):
 
     record_id: int
     created_at: datetime
-    status: str = Field(..., description="granted / capped / ecpm_missing")
+    status: str = Field(
+        ...,
+        description="granted / capped / ecpm_missing / closed_early / too_short",
+    )
     ecpm: str | None = Field(None, description="本次采用的 eCPM 原始值(分/千次展示)")
     ecpm_factor: float | None = Field(None, description="因子1(eCPM 档);非 granted 为空")
     units: int = Field(..., description="折算份数:激励视频恒 1;信息流 = 满 10 秒份数")
@@ -44,7 +47,7 @@ class AdRevenueDaily(BaseModel):
 
     date: str = Field(..., description="北京时间 YYYY-MM-DD")
     impressions: int = Field(..., description="当天展示条数合计")
-    revenue_yuan: float = Field(..., description="当天客户端预估收益合计(元;eCPM 折算)")
+    revenue_yuan: float = Field(..., description="当天客户端有效预估收益合计(元;eCPM 折算)")
     pangle_revenue_yuan: float | None = Field(
         None, description="当天穿山甲后台预估收益(元;GroMore revenue);非全量视图/无数据为空"
     )
@@ -93,7 +96,10 @@ class AdRevenueRow(BaseModel):
     has_impression: bool = Field(..., description="是否有广告展示(信息流逐条展示=True,纯发奖行=False)")
     impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)")
     ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
-    revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0")
+    revenue_yuan: float = Field(
+        ...,
+        description="本次有效展示预估收益(元)= eCPM元 ÷ 1000;纯发奖、激励视频提前关闭/时长不足=0",
+    )
     row_revenue_yuan: float | None = Field(
         None,
         description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
@@ -150,7 +156,7 @@ class AdRevenueReportOut(BaseModel):
     total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
     truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
     total_impressions: int = Field(..., description="全量展示条数合计")
-    total_revenue_yuan: float = Field(..., description="全量客户端预估收益合计(元;eCPM 折算)")
+    total_revenue_yuan: float = Field(..., description="全量客户端有效预估收益合计(元;eCPM 折算)")
     total_pangle_revenue_yuan: float | None = Field(
         None,
         description="全量穿山甲后台预估收益合计(元;GroMore revenue)。穿山甲无用户/类型/场景维度,"

app/admin/schemas/coupon_data.py

@@ -79,10 +79,10 @@ class CouponDataRow(BaseModel):
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
     point_success_count: int | None = Field(
-        None, description="本次成功券点位数(success+already_claimed);无逐券埋点为空"
+        None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
     )
     point_total_count: int | None = Field(
-        None, description="本次尝试券点位数(success+already_claimed+failed,不含 skipped);无逐券埋点为空"
+        None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
     )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(

app/api/v1/compare_record.py

@@ -115,13 +115,22 @@ def list_records(
     db: DbSession,
     limit: int = Query(20, ge=1, le=100),
     cursor: int | None = Query(None, description="上一页末条 id"),
+    ordered: bool | None = Query(
+        None,
+        description="true=只看「已下单」(店名命中本人真实下单)的记录;不传=全部",
+    ),
+    keyword: str | None = Query(
+        None,
+        max_length=64,
+        description="按店名 / 菜名模糊搜索,忽略大小写;空白串等同不传",
+    ),
     include_trace: bool = Query(
         False,
         description="客户端开了本机 agent 调试模式时带 true,放行本人记录的 trace_url",
     ),
 ) -> ComparisonRecordPage:
     items, next_cursor = crud_compare.list_records(
-        db, user.id, limit=limit, cursor=cursor
+        db, user.id, limit=limit, cursor=cursor, ordered=ordered, keyword=keyword
     )
     outs = [ComparisonRecordOut.model_validate(it) for it in items]
     # 权限闸:未开 debug_trace_enabled 的用户不下发 trace_url(列表页「复制调试链接」靠它)。

app/api/v1/coupon.py

@@ -81,7 +81,7 @@ def _record_claims_blocking(
     device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
 ) -> None:
     with SessionLocal() as db:
-        # 取本次 session 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)。
+        # 取本次 session 环境，给每日资产和逐次事件同时打环境标。
         app_env = coupon_repo.session_app_env(db, trace_id)
         coupon_repo.record_claims(db, device_id, user_id, trace_id, results, app_env=app_env)
         # 顺带把本帧「成功平台」并入 coupon_session.platform_success(admin 领券数据 ②整单/③点位成功率;
@@ -176,7 +176,7 @@ async def coupon_step(
 
     resp_json = resp.json()
 
-    # 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
+    # 领券结果沉淀:每日资产 + 逐次事件；中间帧和 done 全量帧均幂等写库。
     # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
     if device_id:
         results = _extract_coupon_results(resp_json)

app/core/config.py

@@ -188,6 +188,13 @@ class Settings(BaseSettings):
         """京东联盟订单查询凭证齐全。"""
         return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET)
 
+    # 美团 + 京东 CPS 订单自动对账：进程内 worker 每天北京时间 05:00 后跑一轮。
+    # 按更新时间回拉近 N 天（重叠窗口防漏单并刷新状态），order_id 幂等更新；手动接口不受影响。
+    CPS_AUTO_RECONCILE_ENABLED: bool = True
+    CPS_AUTO_RECONCILE_RUN_HOUR: int = 5
+    CPS_AUTO_RECONCILE_LOOKBACK_DAYS: int = 3
+    CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC: int = 60
+
     # ===== 微信服务号(网页授权) =====
     # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。
     # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。

app/core/cps_reconcile_worker.py

@@ -0,0 +1,428 @@
+"""美团、京东 CPS 订单每日自动对账任务。
+
+每天北京时间 `CPS_AUTO_RECONCILE_RUN_HOUR`（默认 05:00）后执行一次，按更新时间
+回拉最近若干天订单并复用 admin CPS 仓储层的幂等 upsert。手动对账接口保持独立、不受影响。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import socket
+import time
+from collections.abc import Callable, Iterator
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from uuid import uuid4
+
+from app.admin.repositories import cps as cps_repo
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal, engine
+
+logger = logging.getLogger("shagua.cps_reconcile")
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"
+
+
+def _cn_now() -> datetime:
+    return datetime.now(CN_TZ)
+
+
+def _new_run_id(now: datetime) -> str:
+    return f"{now:%Y%m%d-%H%M%S}-{uuid4().hex[:8]}"
+
+
+def _next_run_at(now: datetime, run_hour: int) -> datetime:
+    scheduled = now.replace(hour=run_hour, minute=0, second=0, microsecond=0)
+    return scheduled if now < scheduled else scheduled + timedelta(days=1)
+
+
+def _trigger_for_run(worker_started_at: datetime, now: datetime, run_hour: int) -> str:
+    if worker_started_at.date() == now.date() and worker_started_at.hour >= run_hour:
+        return "startup_catchup"
+    return "scheduled"
+
+
+def _touch_lock() -> None:
+    with contextlib.suppress(FileNotFoundError):
+        os.utime(_LOCK_PATH, None)
+
+
+@contextlib.contextmanager
+def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
+    """同机多进程保护：同一时间只允许一个 CPS 自动对账 worker 运行。"""
+    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd: int | None = None
+    try:
+        try:
+            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        except FileExistsError:
+            try:
+                age = time.time() - _LOCK_PATH.stat().st_mtime
+            except FileNotFoundError:
+                age = stale_after_sec + 1
+            if age > stale_after_sec:
+                with contextlib.suppress(FileNotFoundError):
+                    _LOCK_PATH.unlink()
+                try:
+                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+                except FileExistsError:
+                    fd = None
+
+        if fd is None:
+            yield False
+            return
+
+        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
+        yield True
+    finally:
+        if fd is not None:
+            os.close(fd)
+            with contextlib.suppress(FileNotFoundError):
+                _LOCK_PATH.unlink()
+
+
+def _empty_result() -> dict:
+    return {
+        "fetched": 0,
+        "inserted": 0,
+        "updated": 0,
+        "pages": 0,
+        "api_requests": 0,
+    }
+
+
+def _platform_log_fields(platform: str, result: dict) -> dict:
+    fields = {
+        "platform": platform,
+        "platform_status": result["status"],
+        "duration_ms": result["duration_ms"],
+    }
+    for key in ("fetched", "inserted", "updated", "pages", "api_requests", "windows"):
+        if key in result:
+            fields[key] = result[key]
+    return fields
+
+
+def _run_platform(
+    *,
+    platform: str,
+    query_time_type: int,
+    common: dict,
+    reconcile: Callable[[], dict],
+) -> tuple[dict, str | None]:
+    started = time.perf_counter()
+    logger.info(
+        "CPS auto reconcile platform started run_id=%s platform=%s",
+        common["run_id"],
+        platform,
+        extra={
+            **common,
+            "event": "cps_reconcile.platform_started",
+            "platform": platform,
+            "query_time_type": query_time_type,
+        },
+    )
+    try:
+        raw_result = reconcile()
+    except Exception as exc:  # noqa: BLE001 - 单平台失败不能阻断另一平台
+        result = {
+            "status": "failed",
+            "duration_ms": round((time.perf_counter() - started) * 1000, 3),
+            "error_type": type(exc).__name__,
+            "error_summary": str(exc)[:500],
+        }
+        logger.exception(
+            "CPS auto reconcile platform failed run_id=%s platform=%s error_type=%s",
+            common["run_id"],
+            platform,
+            type(exc).__name__,
+            extra={
+                **common,
+                "event": "cps_reconcile.platform_failed",
+                **_platform_log_fields(platform, result),
+                "error_type": type(exc).__name__,
+                "error_summary": str(exc)[:500],
+            },
+        )
+        return result, str(exc)
+
+    result = {
+        **raw_result,
+        "status": "success",
+        "duration_ms": round((time.perf_counter() - started) * 1000, 3),
+    }
+    logger.info(
+        "CPS auto reconcile platform completed run_id=%s platform=%s fetched=%s inserted=%s updated=%s",
+        common["run_id"],
+        platform,
+        result.get("fetched", 0),
+        result.get("inserted", 0),
+        result.get("updated", 0),
+        extra={
+            **common,
+            "event": "cps_reconcile.platform_completed",
+            **_platform_log_fields(platform, result),
+        },
+    )
+    return result, None
+
+
+def _skip_platform(platform: str, common: dict) -> dict:
+    result = {
+        **_empty_result(),
+        "status": "skipped",
+        "skipped": "not_configured",
+        "duration_ms": 0.0,
+    }
+    logger.warning(
+        "CPS auto reconcile platform skipped run_id=%s platform=%s reason=not_configured",
+        common["run_id"],
+        platform,
+        extra={
+            **common,
+            "event": "cps_reconcile.platform_skipped",
+            **_platform_log_fields(platform, result),
+            "skip_reason": "not_configured",
+        },
+    )
+    return result
+
+
+def _reconcile_once(
+    now: datetime | None = None,
+    *,
+    run_id: str | None = None,
+    trigger: str = "scheduled",
+) -> dict:
+    """独立拉取美团和京东；单平台异常只记日志，不影响另一平台。"""
+    end = (now or _cn_now()).astimezone(CN_TZ)
+    run_id = run_id or _new_run_id(end)
+    lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
+    start = end - timedelta(days=lookback_days)
+    task_started = time.perf_counter()
+    common = {
+        "run_id": run_id,
+        "trigger": trigger,
+        "scheduled_date": end.date().isoformat(),
+        "app_env": settings.APP_ENV,
+        "db_dialect": engine.dialect.name,
+        "hostname": socket.gethostname(),
+        "pid": os.getpid(),
+        "window_start": start.isoformat(),
+        "window_end": end.isoformat(),
+        "lookback_days": lookback_days,
+    }
+    result = {
+        "run_id": run_id,
+        "trigger": trigger,
+        "window_start": start.isoformat(),
+        "window_end": end.isoformat(),
+        "meituan": None,
+        "jd": None,
+        "errors": {},
+    }
+    logger.info(
+        "CPS auto reconcile started run_id=%s trigger=%s window=%s..%s",
+        run_id,
+        trigger,
+        result["window_start"],
+        result["window_end"],
+        extra={**common, "event": "cps_reconcile.started"},
+    )
+
+    if settings.mt_cps_configured:
+        def reconcile_meituan() -> dict:
+            with SessionLocal() as db:
+                return cps_repo.reconcile_orders(
+                    db,
+                    start_time=int(start.timestamp()),
+                    end_time=int(end.timestamp()),
+                    query_time_type=2,
+                    audit_context=common,
+                )
+
+        result["meituan"], error = _run_platform(
+            platform="meituan",
+            query_time_type=2,
+            common=common,
+            reconcile=reconcile_meituan,
+        )
+        if error is not None:
+            result["errors"]["meituan"] = error
+    else:
+        result["meituan"] = _skip_platform("meituan", common)
+
+    _touch_lock()
+    if settings.jd_union_configured:
+        def reconcile_jd() -> dict:
+            with SessionLocal() as db:
+                return cps_repo.reconcile_jd_orders(
+                    db,
+                    start_time=start,
+                    end_time=end,
+                    query_time_type=3,
+                    audit_context=common,
+                )
+
+        result["jd"], error = _run_platform(
+            platform="jd",
+            query_time_type=3,
+            common=common,
+            reconcile=reconcile_jd,
+        )
+        if error is not None:
+            result["errors"]["jd"] = error
+    else:
+        result["jd"] = _skip_platform("jd", common)
+
+    successes = sum(
+        platform_result.get("status") == "success"
+        for platform_result in (result["meituan"], result["jd"])
+    )
+    if result["errors"]:
+        task_status = "partial_success" if successes else "failed"
+    else:
+        task_status = "success"
+    completed_at = _cn_now()
+    duration_ms = round((time.perf_counter() - task_started) * 1000, 3)
+    next_run_at = _next_run_at(
+        completed_at,
+        min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23),
+    ).isoformat()
+    result.update({
+        "status": task_status,
+        "duration_ms": duration_ms,
+        "manual_retry_required": bool(result["errors"]),
+        "next_run_at": next_run_at,
+    })
+    completion_fields = {
+        **common,
+        "event": "cps_reconcile.completed",
+        "task_status": task_status,
+        "duration_ms": duration_ms,
+        "manual_retry_required": result["manual_retry_required"],
+        "failed_platforms": sorted(result["errors"]),
+        "next_run_at": next_run_at,
+        "meituan_result": result["meituan"],
+        "jd_result": result["jd"],
+    }
+    log_method = logger.info if task_status == "success" else logger.warning
+    log_method(
+        "CPS auto reconcile completed run_id=%s status=%s duration_ms=%s failed_platforms=%s next_run_at=%s",
+        run_id,
+        task_status,
+        duration_ms,
+        ",".join(sorted(result["errors"])) or "-",
+        next_run_at,
+        extra=completion_fields,
+    )
+
+    return result
+
+
+def _should_run(last_run: date | None, now: datetime, run_hour: int) -> bool:
+    return last_run != now.date() and now.hour >= run_hour
+
+
+async def _run_loop() -> None:
+    interval = max(30, int(settings.CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC))
+    run_hour = min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23)
+    lock_stale_after = max(interval * 3, 1800)
+    with _single_instance_lock(lock_stale_after) as lock_acquired:
+        if not lock_acquired:
+            logger.warning(
+                "CPS auto reconcile worker skipped: another worker owns lock",
+                extra={
+                    "event": "cps_reconcile.worker_skipped",
+                    "skip_reason": "lock_not_acquired",
+                    "pid": os.getpid(),
+                },
+            )
+            return
+        await _run_locked_loop(interval, run_hour)
+
+
+async def _run_locked_loop(interval: int, run_hour: int) -> None:
+    worker_started_at = _cn_now()
+    logger.info(
+        "CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
+        run_hour,
+        interval,
+        settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
+        extra={
+            "event": "cps_reconcile.worker_started",
+            "pid": os.getpid(),
+            "hostname": socket.gethostname(),
+            "app_env": settings.APP_ENV,
+            "db_dialect": engine.dialect.name,
+            "run_hour": run_hour,
+            "interval_sec": interval,
+            "lookback_days": settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
+            "next_run_at": _next_run_at(worker_started_at, run_hour).isoformat(),
+        },
+    )
+    last_run: date | None = None
+    try:
+        while True:
+            run_id: str | None = None
+            try:
+                _touch_lock()
+                now = _cn_now()
+                if _should_run(last_run, now, run_hour):
+                    run_id = _new_run_id(now)
+                    trigger = _trigger_for_run(worker_started_at, now, run_hour)
+                    await asyncio.to_thread(
+                        _reconcile_once,
+                        now,
+                        run_id=run_id,
+                        trigger=trigger,
+                    )
+                    last_run = now.date()
+            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
+                logger.exception(
+                    "CPS auto reconcile unexpected worker error",
+                    extra={
+                        "event": "cps_reconcile.unexpected_failed",
+                        "run_id": run_id or "unassigned",
+                        "pid": os.getpid(),
+                    },
+                )
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info(
+            "CPS auto reconcile worker stopped",
+            extra={"event": "cps_reconcile.worker_stopped", "pid": os.getpid()},
+        )
+        raise
+
+
+def start_cps_reconcile_worker() -> asyncio.Task | None:
+    if not settings.CPS_AUTO_RECONCILE_ENABLED:
+        logger.info(
+            "CPS auto reconcile disabled",
+            extra={
+                "event": "cps_reconcile.worker_skipped",
+                "skip_reason": "disabled",
+            },
+        )
+        return None
+    if not settings.mt_cps_configured and not settings.jd_union_configured:
+        logger.warning(
+            "CPS auto reconcile not started: Meituan and JD credentials are missing",
+            extra={
+                "event": "cps_reconcile.worker_skipped",
+                "skip_reason": "all_platform_credentials_missing",
+            },
+        )
+        return None
+    return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")
+
+
+async def stop_cps_reconcile_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/main.py

@@ -43,6 +43,10 @@ from app.api.v1.user import router as user_router
 from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
 from app.core.config import settings
+from app.core.cps_reconcile_worker import (
+    start_cps_reconcile_worker,
+    stop_cps_reconcile_worker,
+)
 from app.core.daily_exchange_worker import (
     start_daily_exchange_worker,
     stop_daily_exchange_worker,
@@ -89,6 +93,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     except Exception:  # noqa: BLE001
         logger.exception("reverse_geocoder 预热失败(城市反查将在首个请求时懒加载)")
     reconcile_task = start_withdraw_reconcile_worker()
+    cps_reconcile_task = start_cps_reconcile_worker()
     heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
@@ -98,6 +103,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     finally:
         await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
+        await stop_cps_reconcile_worker(cps_reconcile_task)
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)

app/models/__init__.py

@@ -21,6 +21,7 @@ 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,
     CouponDailyCompletion,
     CouponPromptEngagement,

app/models/comparison.py

@@ -45,6 +45,10 @@ class ComparisonRecord(Base):
         # 首页轮播 / 省钱战绩聚合都按 status='success' 过滤 + created_at 近期排序;
         # 复合索引避免随数据量增大退化成全表扫(单列 created_at 索引不含 status)。
         Index("ix_comparison_status_created", "status", "created_at"),
+        # C 端「我的比价记录」列表:WHERE user_id=? ORDER BY created_at DESC, id DESC LIMIT n。
+        # 单列 user_id 索引只能过滤,排序仍要把该用户全部记录取出来排一遍;这条复合索引的**反向扫**
+        # 恰好等于 (created_at DESC, id DESC),PG 直接取前 n 条、免排序。列序不能动。
+        Index("ix_comparison_user_created", "user_id", "created_at", "id"),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

app/models/coupon_state.py

@@ -96,6 +96,40 @@ class CouponClaimRecord(Base):
         )
 
 
+class CouponClaimEvent(Base):
+    """一次领券任务中的单券结果，按 ``(trace_id, coupon_id)`` 幂等。"""
+
+    __tablename__ = "coupon_claim_event"
+    __table_args__ = (
+        UniqueConstraint(
+            "trace_id", "coupon_id",
+            name="uq_coupon_claim_event_trace_coupon",
+        ),
+        Index("ix_coupon_claim_event_date_env", "claim_date", "app_env"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    trace_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+    coupon_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    claim_date: Mapped[date] = mapped_column(Date, nullable=False)
+    status: Mapped[str] = mapped_column(String(24), nullable=False)
+    app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
+    vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
+    coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    extra: 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 CouponDailyCompletion(Base):
     """按 (device, 自然日) 记"今天是否已跑完整轮领券(到 done 帧)"——首页置灰源。
 

app/repositories/comparison.py

@@ -7,8 +7,8 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import func, select
-from sqlalchemy.orm import Session
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session, defer
 
 from app.core.rewards import CN_TZ
 from app.models.ad_feed_reward import AdFeedRewardRecord
@@ -375,19 +375,48 @@ def harvest_abort(
     return rec
 
 
-def _ordered_shop_names(db: Session, user_id: int) -> set[str]:
-    """该用户「真实下单」(source='compare')覆盖到的店名集合,用来给比价记录打「已下单」。
+def _ordered_shop_name_select(user_id: int):
+    """该用户「真实下单」(source='compare')覆盖到的店名 select,给「已下单」筛选当子查询。
+
+    口径与 [_ordered_shop_names] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates
+    反查打标;这边是**分页之前**就要过滤,拿不到 candidates,只能整段下推成子查询。
+    没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的店名可能上千,展开会撞 SQLite
+    的绑定变量上限,而且又变回了那个「随下单量线性变慢」的老写法。
+    """
+    return select(SavingsRecord.shop_name).where(
+        SavingsRecord.user_id == user_id,
+        SavingsRecord.source == "compare",
+        SavingsRecord.shop_name.is_not(None),
+    )
+
+
+def _like_escape(kw: str) -> str:
+    """转义 LIKE 通配符(百分号 / 下划线 / 反斜杠),让用户输入只按字面量匹配(配合 escape 参数)。
+
+    不转义的话搜一个「%」就等于把整表拉回来。
+    """
+    return kw.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+def _ordered_shop_names(db: Session, user_id: int, candidates: set[str]) -> set[str]:
+    """[candidates] 里哪些店名被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。
 
     只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id,
     只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。
     语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。
+
+    ⚠️ 只查**本页出现过的店名**(candidates ≤ limit 条),不再把该用户全部下单店名捞回内存:
+    老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回
+    (避免 IN () 非法)。
     """
+    if not candidates:
+        return set()
     rows = db.execute(
         select(SavingsRecord.shop_name).where(
             SavingsRecord.user_id == user_id,
             SavingsRecord.source == "compare",
-            SavingsRecord.shop_name.is_not(None),
-        )
+            SavingsRecord.shop_name.in_(candidates),
+        ).distinct()
     ).scalars().all()
     return {s for s in rows if s}
 
@@ -415,17 +444,60 @@ def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[
     return {tid: int(coin) for tid, coin in rows if tid}
 
 
+# 列表出参(ComparisonRecordOut)根本不读、但 select(ORM) 默认会一并捞回来的重型 JSON 列:
+#   - raw_payload:done.params 上报体全量,**每条记录都有**(harvest 与 POST 两条写路径都落)。
+#     单条几 KB~几十 KB,一页 50 条就是稳定几百 KB~几 MB 的白读 + 白反序列化。
+#   - llm_calls:每次 LLM 调用的 input_messages + output 全文。只有走老客户端 POST /compare/record
+#     的记录才有(_backfill_llm_calls 回填;harvest 路径不落),但有的时候单条就能到 MB 级 —— 一页里
+#     混进几条这种记录,整个请求就被它们拖住。
+#   - llm_price_snapshot:逐模型单价快照,同样只在回填时落。
+# 三列全部读出来再被 pydantic 丢掉,是「比价记录/全部记录」页慢的主要来源。
+# ⚠️ defer 的列一旦在别处被读到会触发**逐行**懒加载(N+1);列表这条链路(ComparisonRecordOut
+# 不声明这三个字段 → 不会 getattr 到)是安全的。详情接口 get_record 不 defer,raw_payload 照常返回。
+_LIST_DEFERRED = (
+    ComparisonRecord.raw_payload,
+    ComparisonRecord.llm_calls,
+    ComparisonRecord.llm_price_snapshot,
+)
+
+
 def list_records(
     db: Session,
     user_id: int,
     *,
     limit: int = 20,
     cursor: int | None = None,
+    ordered: bool | None = None,
+    keyword: str | None = None,
 ) -> tuple[list[ComparisonRecord], int | None]:
     """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。"""
-    stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id)
+    stmt = (
+        select(ComparisonRecord)
+        .where(ComparisonRecord.user_id == user_id)
+        .options(*(defer(col) for col in _LIST_DEFERRED))
+    )
     if cursor is not None:
         stmt = stmt.where(ComparisonRecord.id < cursor)
+    # 「已下单」tab 与搜索框的过滤都下推到这里,不能留给客户端对整页结果 filter ——
+    # 分页之后一页里可能一条都不命中,列表看着就是空的/卡住的,得翻很多页才蹦出一条。
+    if ordered:
+        stmt = stmt.where(
+            ComparisonRecord.store_name.in_(_ordered_shop_name_select(user_id))
+        )
+    kw = (keyword or "").strip()
+    if kw:
+        # product_names 是写路径从 items[].name 派生的普通文本列(items 本身是 JSON,SQLite 下
+        # 中文被 ensure_ascii 转义,没法直接 LIKE)—— 搜「菜名」靠的就是它。
+        # ilike:PG 原生 ILIKE,SQLite 渲染成 lower() LIKE lower(),两边都忽略大小写。
+        pattern = f"%{_like_escape(kw)}%"
+        stmt = stmt.where(
+            or_(
+                ComparisonRecord.store_name.ilike(pattern, escape="\\"),
+                ComparisonRecord.product_names.ilike(pattern, escape="\\"),
+            )
+        )
+    # 排序与 ix_comparison_user_created(user_id, created_at, id)对齐 —— DESC/DESC 正好是该索引的
+    # 反向扫,PG 免排序直接取前 limit 条。改排序方向前先想清楚索引还吃不吃得上。
     stmt = stmt.order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()).limit(limit)
 
     items = list(db.execute(stmt).scalars().all())
@@ -433,7 +505,10 @@ def list_records(
 
     # 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
     # ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
-    ordered_shops = _ordered_shop_names(db, user_id)
+    page_shops = {it.store_name for it in items if it.store_name}
+    # ordered=True 时上面已按同一口径(_ordered_shop_name_select)筛过,本页必然全是已下单,
+    # 省掉这次反查;其余情况照旧按本页店名反查 savings。
+    ordered_shops = page_shops if ordered else _ordered_shop_names(db, user_id, page_shops)
     # 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。
     ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items])
     for it in items:

app/repositories/coupon_state.py

@@ -14,6 +14,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.models.coupon_state import (
+    CouponClaimEvent,
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
@@ -164,11 +165,12 @@ def record_claims(
     results: list[dict],
     app_env: str | None = None,
 ) -> int:
-    """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。
+    """一批券领取结果同时写入每日资产表和逐次事件表。
 
     results 单项取自 pricebot 的 last_coupon_result / done.coupon_results,识别字段:
     coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)。
-    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
+    - CouponClaimRecord 按 (device, coupon_id, 今天) 幂等，供每日资产口径使用。
+    - CouponClaimEvent 按 (trace_id, coupon_id) 幂等，供 admin 逐场统计使用。
     """
     today = today_cn()
     written = 0
@@ -208,6 +210,41 @@ def record_claims(
                 claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
                 extra=r,
             ))
+        if trace_id:
+            event = db.execute(
+                select(CouponClaimEvent).where(
+                    CouponClaimEvent.trace_id == trace_id,
+                    CouponClaimEvent.coupon_id == coupon_id,
+                )
+            ).scalar_one_or_none()
+            if event is not None:
+                event.device_id = device_id
+                event.status = status
+                event.reason = r.get("reason")
+                event.vendor = r.get("vendor")
+                event.coupon_name = r.get("name")
+                event.extra = r
+                if user_id is not None:
+                    event.user_id = user_id
+                if count is not None:
+                    event.claimed_count = count
+                if app_env is not None:
+                    event.app_env = app_env
+            else:
+                db.add(CouponClaimEvent(
+                    trace_id=trace_id,
+                    device_id=device_id,
+                    user_id=user_id,
+                    coupon_id=coupon_id,
+                    claim_date=today,
+                    status=status,
+                    app_env=app_env,
+                    vendor=r.get("vendor"),
+                    coupon_name=r.get("name"),
+                    claimed_count=count,
+                    reason=r.get("reason"),
+                    extra=r,
+                ))
         written += 1
     if written == 0:
         return 0

deploy/nginx/app-api.shaguabijia.com.conf

@@ -25,6 +25,18 @@ server {
     # (纯文字反馈体积小、不受影响 → 呈现为「时好时坏」)。根治仍需客户端上传前压缩。
     client_max_body_size 32m;
 
+    # JSON 响应压缩。nginx 默认 gzip off,且就算 on 了 gzip_types 也只含 text/html、
+    # gzip_proxied 默认 off(反代来的响应一律不压)—— 三个默认值凑一起 = 我们所有接口都在裸奔。
+    # 比价记录列表这种一次 50 条、字段名 + 中文店名/菜名高度重复的 JSON,gzip 压缩比稳定在 8~10 倍
+    # (几百 KB → 几十 KB),弱网下省的就是首屏那几秒。
+    # 只压 JSON:APK 直链(/media/shaguabijia.apk)、图片本身已是压缩格式,再压纯浪费 CPU。
+    gzip              on;
+    gzip_proxied      any;      # 反代响应也压(默认 off = 对我们这套反代等于没开)
+    gzip_types        application/json;
+    gzip_min_length   1024;     # 小响应压了反而更大(gzip 头开销),不值当
+    gzip_comp_level   5;        # 5 是体积/CPU 的常用折中点,再往上收益递减
+    gzip_vary         on;       # 给 CDN/中间缓存正确按 Accept-Encoding 分桶
+
     location / {
         proxy_pass http://127.0.0.1:8770;
         proxy_http_version 1.1;

docs/api/admin/admin-audit-logs.md

@@ -1,6 +1,6 @@
 # GET /admin/api/audit-logs — 审计日志(谁改了什么,游标分页)
 
-> 所属:Admin·Audit 组（前缀 `/admin/api/audit-logs`） | 鉴权:Bearer admin_token（角色:任意已登录 admin） | [← 返回 API 索引](../README.md)
+> 所属:Admin·Audit 组（前缀 `/admin/api/audit-logs`） | 鉴权:Bearer admin_token + `audit-logs` 页面权限 | [← 返回 API 索引](../README.md)
 
 ## 入参（query）
 | 字段 | 类型 | 必填 | 默认 | 说明 |
@@ -29,7 +29,8 @@
 
 ## 错误码
 - `401` 未带 admin token / token 无效或过期 / 管理员被禁用
+- `403` 当前管理员没有 `audit-logs` 页面权限
 
 ## 说明
-- 整组(`/admin/api/audit-logs`)守卫为 `get_current_admin`,任意已登录 admin 均可查看,无角色限制。
+- 整组(`/admin/api/audit-logs`)守卫为 `require_page("audit-logs")`，默认仅超级管理员和技术角色可查看，也可由超管给自定义角色授权。
 - 审计日志只增不改不删,任何写操作经 `write_audit` 落一条。数据表见 [admin_audit_log](../database/admin_audit_log.md)。

docs/api/admin/admin-device-liveness.md

@@ -1,6 +1,6 @@
 # /admin/api/device-liveness — 设备存活监控(#80)
 
-> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin | 表 [device_liveness](../../database/device_liveness.md) | [← 返回 API 索引](../README.md)
+> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin + `device-liveness` 页面权限 | 表 [device_liveness](../../database/device_liveness.md) | [← 返回 API 索引](../README.md)
 
 无障碍保护存活的后台视角:哪些设备开过保护(`ever_protected`)、现在在线还是掉线(心跳超时,#107 起阈值 1 小时)、首次开启时间(`first_protected_at`)。
 
@@ -13,3 +13,4 @@
 
 ## 说明
 - 「在线」= `last_heartbeat_at` 距今 < 超时阈值;掉线召回链路(worker 置 `kill_alert_pending` → 客户端 pull)见表文档。
+- 无 `device-liveness` 页面权限时返回 `403`。

docs/api/admin/admin-event-logs.md

@@ -1,6 +1,6 @@
 # /admin/api/event-logs — 埋点日志(#83)
 
-> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin | 表 [analytics_event](../../database/analytics_event.md) | [← 返回 API 索引](../README.md)
+> 所属:Admin 子应用(前缀 `/admin/api`) | 鉴权:admin + `event-logs` 页面权限 | 表 [analytics_event](../../database/analytics_event.md) | [← 返回 API 索引](../README.md)
 
 客户端埋点(`POST /api/v1/analytics/events` 批量上报)的后台检索页。
 
@@ -13,3 +13,4 @@
 ## 说明
 - 纯只读;无聚合报表(要分析导出后自己算)。
 - 时间轴用 `client_ts`(事件真实发生时刻),入库时间受客户端攒批影响。
+- 无 `event-logs` 页面权限时返回 `403`。

docs/api/compare/compare-records.md

@@ -10,6 +10,12 @@
 |---|---|---|---|---|
 | `limit` | int | ❌ | 20 | 1–100 |
 | `cursor` | int | ❌ | null | 上一页末条 `id`,首页不传 |
+| `ordered` | bool | ❌ | null | `true`=只出「已下单」(店名命中本人真实下单)的记录;不传=不筛 |
+| `keyword` | string | ❌ | null | 按店名 / 菜名模糊搜索,忽略大小写,≤64 字符;纯空白等同不传 |
+| `include_trace` | bool | ❌ | false | 客户端开了本机 agent 调试模式时带 `true`,放行**本人**记录的 `trace_url` |
+
+`ordered` / `keyword` 都在服务端过滤后再分页,客户端不要拿一页结果自己 filter ——
+分页之后一页里可能一条都不命中,列表会看着像空的。
 
 ## 出参
 响应 `200`：`{ items: ComparisonRecordOut[], next_cursor: int|null }`（分页见 [索引#游标分页约定](./README.md#游标分页约定)）
@@ -38,6 +44,9 @@
 | `items` | object[] | 下单菜品 `{name, qty, specs?}` |
 | `comparison_results` | object[] | 逐平台对比（price 单位元，已按 rank 升序） |
 | `skipped_dish_names` | string[] | 被跳过的菜名 |
+| `ordered` | bool | 「已下单」店级标记:店名命中本人 `source='compare'` 的下单记录即 `true`。**瞬态字段,不在表里**,每次查询现算 |
+| `ad_coins_earned` | int | 本次比价看信息流广告实发的金币(按 `trace_id` 聚合)。同为瞬态字段 |
+| `trace_url` | string \| null | pricebot 调试链接。未开 `debug_trace_enabled` 且未带 `include_trace=true` 时为 `null` |
 | `created_at` | datetime | 时间 |
 
 ## 错误

scripts/seed_coupon_session_mock.py

@@ -0,0 +1,197 @@
+"""生成 admin「领券记录」本地联调数据。
+
+用法:
+    python scripts/seed_coupon_session_mock.py
+
+脚本只清理 ``mock-coupon-repeat-*`` 前缀的数据并重新生成。打开后台「领券记录」，
+日期选今天；分别切换 prod/dev，可验证同一设备同一天多次领券仍各自显示正确分数。
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import delete, select
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from app.db.session import SessionLocal, engine  # noqa: E402
+from app.models.coupon_state import (  # noqa: E402
+    CouponClaimEvent,
+    CouponClaimRecord,
+    CouponSession,
+)
+from app.models.user import User  # noqa: E402
+from app.repositories.coupon_state import record_claims, today_cn  # noqa: E402
+
+PREFIX = "mock-coupon-repeat-"
+CN_TZ = ZoneInfo("Asia/Shanghai")
+DEVICE_REPEAT = f"{PREFIX}device"
+DEVICE_CONTROL = f"{PREFIX}control-device"
+PHONE = "19900009001"
+USERNAME = "80000009001"
+PLATFORM_ELAPSED = {
+    "meituan-waimai": 46_800,
+    "taobao-shanguang": 19_500,
+    "jd-waimai": 24_000,
+}
+
+FIRST_RESULTS = [
+    {"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "success"},
+    {"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "success"},
+    {"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"},
+    {"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "failed", "reason": "模拟失败"},
+    {"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"},
+    {"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"},
+    {"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "success"},
+    {"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"},
+]
+
+SECOND_RESULTS = [
+    {"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "skipped", "reason": "模拟跳过，不计分母"},
+    {"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "already_claimed"},
+    {"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"},
+    {"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "success"},
+    {"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"},
+    {"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"},
+    {"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "already_claimed"},
+    {"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"},
+]
+
+
+def _started_at(hour: int, minute: int) -> datetime:
+    local = datetime.combine(today_cn(), datetime.min.time()).replace(
+        hour=hour, minute=minute, tzinfo=CN_TZ
+    )
+    return local.astimezone(UTC)
+
+
+def _session(
+    *,
+    trace_id: str,
+    device_id: str,
+    user_id: int,
+    app_env: str,
+    hour: int,
+    minute: int,
+    status: str = "completed",
+    elapsed_ms: int | None = 91_900,
+    platform_elapsed: dict[str, int] | None = None,
+) -> CouponSession:
+    started_at = _started_at(hour, minute)
+    return CouponSession(
+        trace_id=trace_id,
+        device_id=device_id,
+        user_id=user_id,
+        status=status,
+        app_env=app_env,
+        platforms=[],
+        origin_package=None,
+        device_model="Mock Phone",
+        rom="MockOS 1",
+        started_at=started_at,
+        started_date=today_cn(),
+        finished_at=started_at if status != "started" else None,
+        elapsed_ms=elapsed_ms,
+        platform_elapsed=platform_elapsed,
+        platform_success=(
+            ["meituan-waimai", "taobao-shanguang", "jd-waimai"]
+            if status == "completed" else None
+        ),
+        claimed_count=7 if status == "completed" else 0,
+    )
+
+
+def main() -> None:
+    # 本地旧库 Alembic 版本链可能未同步；仅为联调补建新事件表，正式环境仍走 migration。
+    CouponClaimEvent.__table__.create(bind=engine, checkfirst=True)
+    with SessionLocal() as db:
+        db.execute(delete(CouponClaimEvent).where(
+            CouponClaimEvent.trace_id.startswith(PREFIX)
+        ))
+        db.execute(delete(CouponClaimRecord).where(
+            CouponClaimRecord.device_id.startswith(PREFIX)
+        ))
+        db.execute(delete(CouponSession).where(
+            CouponSession.trace_id.startswith(PREFIX)
+        ))
+        user = db.execute(select(User).where(User.phone == PHONE)).scalar_one_or_none()
+        if user is None:
+            user = User(
+                phone=PHONE,
+                username=USERNAME,
+                register_channel="sms",
+                nickname="领券重复测试",
+            )
+            db.add(user)
+            db.flush()
+
+        first_trace = f"{PREFIX}dev-first"
+        second_trace = f"{PREFIX}prod-second"
+        abandoned_trace = f"{PREFIX}prod-abandoned"
+        control_trace = f"{PREFIX}prod-control"
+        db.add_all([
+            _session(
+                trace_id=first_trace,
+                device_id=DEVICE_REPEAT,
+                user_id=user.id,
+                app_env="dev",
+                hour=10,
+                minute=0,
+                platform_elapsed=PLATFORM_ELAPSED,
+            ),
+            _session(
+                trace_id=second_trace,
+                device_id=DEVICE_REPEAT,
+                user_id=user.id,
+                app_env="prod",
+                hour=15,
+                minute=0,
+                platform_elapsed=PLATFORM_ELAPSED,
+            ),
+            _session(
+                trace_id=abandoned_trace,
+                device_id=DEVICE_REPEAT,
+                user_id=user.id,
+                app_env="prod",
+                hour=16,
+                minute=0,
+                status="abandoned",
+                elapsed_ms=21_500,
+                platform_elapsed={"meituan-waimai": 20_500},
+            ),
+            _session(
+                trace_id=control_trace,
+                device_id=DEVICE_CONTROL,
+                user_id=user.id,
+                app_env="prod",
+                hour=17,
+                minute=0,
+                platform_elapsed=PLATFORM_ELAPSED,
+            ),
+        ])
+        db.commit()
+
+        record_claims(
+            db, DEVICE_REPEAT, user.id, first_trace, FIRST_RESULTS, app_env="dev"
+        )
+        record_claims(
+            db, DEVICE_REPEAT, user.id, second_trace, SECOND_RESULTS, app_env="prod"
+        )
+        record_claims(
+            db, DEVICE_CONTROL, user.id, control_trace, FIRST_RESULTS, app_env="prod"
+        )
+
+    print(f"已生成 {today_cn()} 的领券 mock 数据。")
+    print("筛选用户 19900009001。")
+    print("prod 应有：7/7（100.0%）、-、7/8（87.5%）三条。")
+    print("dev 应有：7/8（87.5%）一条。")
+
+
+if __name__ == "__main__":
+    main()

tests/test_admin_ad_revenue_scope.py

@@ -9,9 +9,11 @@ from app.admin.repositories import ad_revenue
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_pangle_revenue import AdPangleDailyRevenue
+from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
 
 REPORT_DATE = "2040-02-03"
+PLAYBACK_DATE = "2040-02-04"
 
 
 def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -> None:
@@ -127,3 +129,82 @@ def test_business_scope_filters_client_and_pangle_by_env_and_code(monkeypatch) -
         db.execute(delete(User).where(User.phone == "18800009991"))
         db.commit()
         db.close()
+
+
+def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
+    db = SessionLocal()
+    phone = "18800009992"
+    sessions = {
+        "closed_early": "rv-zero-closed",
+        "too_short": "rv-zero-short",
+        "capped": "rv-keep-capped",
+        "granted": "rv-keep-granted",
+    }
+    try:
+        user = User(phone=phone, username="29999999992", register_channel="sms")
+        db.add(user)
+        db.flush()
+
+        for index, (status, session_id) in enumerate(sessions.items(), start=1):
+            created_at = datetime(2040, 2, 4, index, tzinfo=UTC)
+            db.add(AdEcpmRecord(
+                user_id=user.id,
+                ad_type="reward_video",
+                ad_session_id=session_id,
+                app_env="prod",
+                our_code_id="prod-reward",
+                ecpm_raw="10000",
+                report_date=PLAYBACK_DATE,
+                created_at=created_at,
+            ))
+            db.add(AdRewardRecord(
+                trans_id=f"{session_id}-trans",
+                user_id=user.id,
+                coin=0,
+                status=status,
+                reward_scene="reward_video",
+                ad_session_id=session_id,
+                app_env="prod",
+                our_code_id="prod-reward",
+                ecpm_raw="10000",
+                reward_date=PLAYBACK_DATE,
+                created_at=created_at,
+            ))
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=PLAYBACK_DATE,
+            date_to=PLAYBACK_DATE,
+            user_id=user.id,
+            ad_type="reward_video",
+            app_env="prod",
+            revenue_scope="all",
+            granularity="hour",
+        )
+
+        revenue_by_status = {row["status"]: row["revenue_yuan"] for row in result["items"]}
+        assert revenue_by_status == {
+            "closed_early": 0.0,
+            "too_short": 0.0,
+            "capped": 0.1,
+            "granted": 0.1,
+        }
+        assert result["total_impressions"] == 4
+        assert result["total_revenue_yuan"] == 0.2
+        assert len(result["daily"]) == 1
+        assert result["daily"][0]["date"] == PLAYBACK_DATE
+        assert result["daily"][0]["impressions"] == 4
+        assert result["daily"][0]["revenue_yuan"] == 0.2
+        assert sum(row["revenue_yuan"] for row in result["hourly"]) == 0.2
+        assert result["type_stats"]["reward_video"] == {
+            "impressions": 4,
+            "revenue_yuan": 0.2,
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(AdRewardRecord).where(AdRewardRecord.reward_date == PLAYBACK_DATE))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == PLAYBACK_DATE))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()

tests/test_admin_roles.py

@@ -44,17 +44,78 @@ def operator_token() -> str:
     return _token("r_operator", "operator")
 
 
+@pytest.fixture()
+def tech_token() -> str:
+    return _token("r_tech", "tech")
+
+
 def _auth(t: str) -> dict:
     return {"Authorization": f"Bearer {t}"}
 
 
 def test_super_pages_all_operator_limited(admin_client, super_token, operator_token) -> None:
     su = admin_client.get("/admin/api/auth/me", headers=_auth(super_token)).json()
-    assert "admins" in su["pages"] and "dashboard" in su["pages"]  # 超管全页
+    assert "admins" in su["pages"] and "analytics-health" in su["pages"]  # 超管全页
     op = admin_client.get("/admin/api/auth/me", headers=_auth(operator_token)).json()
     assert "dashboard" in op["pages"] and "admins" not in op["pages"]  # 运营看不到管理员页
 
 
+def test_monitoring_audit_catalog_and_api_permissions(
+    admin_client, super_token, operator_token, tech_token
+) -> None:
+    catalog = admin_client.get(
+        "/admin/api/roles/catalog", headers=_auth(super_token)
+    ).json()
+    monitoring = next(group for group in catalog if group["group"] == "监控审计")
+    assert [page["key"] for page in monitoring["pages"]] == [
+        "device-liveness", "analytics-health", "event-logs", "audit-logs",
+    ]
+
+    # 运营默认只能查设备存活，不能绕过导航直调技术/审计接口。
+    assert admin_client.get(
+        "/admin/api/device-liveness/stats", headers=_auth(operator_token)
+    ).status_code == 200
+    for path in (
+        "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
+        "/admin/api/event-logs",
+        "/admin/api/audit-logs",
+    ):
+        assert admin_client.get(path, headers=_auth(operator_token)).status_code == 403
+
+    # 技术角色默认拥有监控审计组全部四项权限。
+    for path in (
+        "/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",
+        "/admin/api/audit-logs",
+    ):
+        assert admin_client.get(path, headers=_auth(tech_token)).status_code == 200
+
+
+def test_custom_admin_api_permission_uses_pages_override(admin_client) -> None:
+    db = SessionLocal()
+    try:
+        admin = admin_repo.get_by_username(db, "r_monitoring_custom")
+        if admin is None:
+            admin = admin_repo.create_admin(
+                db, username="r_monitoring_custom", password="pass1234", role="custom"
+            )
+        admin.password_hash = hash_password("pass1234")
+        admin.role = "custom"
+        admin.pages_override = ["event-logs"]
+        admin.status = "active"
+        db.commit()
+    finally:
+        db.close()
+
+    token = admin_client.post(
+        "/admin/api/auth/login",
+        json={"username": "r_monitoring_custom", "password": "pass1234"},
+    ).json()["access_token"]
+    assert admin_client.get("/admin/api/event-logs", headers=_auth(token)).status_code == 200
+    assert admin_client.get("/admin/api/audit-logs", headers=_auth(token)).status_code == 403
+
+
 def test_roles_endpoints_super_only(admin_client, super_token, operator_token) -> None:
     assert admin_client.get("/admin/api/roles", headers=_auth(super_token)).status_code == 200
     assert admin_client.get("/admin/api/roles", headers=_auth(operator_token)).status_code == 403
@@ -123,7 +184,7 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
     assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
     assert set(roles["tech"]["pages"]) == {
-        "dashboard", "device-liveness", "config", "ad-revenue", "huawei-review",
+        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
         "event-logs", "audit-logs",
     }
 

tests/test_compare_record.py

@@ -222,6 +222,210 @@ def test_stats_compare_count_and_saved(client) -> None:
     assert s2["compare_count"] == 2  # 仍 2(failed 不计)
 
 
+def test_records_ordered_flag(client) -> None:
+    """「已下单」店级标记:店名命中该用户 source='compare' 的下单记录才 True。
+
+    覆盖 list_records 只按**本页店名**反查 savings 的写法(原来是把该用户全部下单店名捞回内存
+    再取交集,随下单量线性变慢)——两种写法结果必须一致,故这里按店名逐条断言。
+    """
+    token = _login(client, "13800002010")
+
+    # 两条比价记录:一条海底捞(稍后会有对应下单),一条没下过单的店
+    client.post("/api/v1/compare/record", json=_food_payload("ord-1"), headers=_auth(token))
+    other = _food_payload("ord-2")
+    other["store_name"] = "没下过单的店"
+    client.post("/api/v1/compare/record", json=other, headers=_auth(token))
+
+    # 下单前:两条都不该带「已下单」
+    items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
+    assert {it["store_name"]: it["ordered"] for it in items} == {
+        "海底捞(朝阳店)": False,
+        "没下过单的店": False,
+    }
+
+    # 对海底捞真实下单一笔(order/report 写 source='compare' 的 savings_record)
+    r = client.post(
+        "/api/v1/order/report",
+        json={
+            "client_event_id": "evt-ordered-flag",
+            "platform": "美团",
+            "platform_package": "com.sankuai.meituan",
+            "pay_channel": "wechat",
+            "compared_price_cents": 12350,
+            "paid_amount_cents": 12350,
+            "shop_name": "海底捞(朝阳店)",
+            "original_price_cents": 12850,
+        },
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+
+    # 下单后:只有同店名那条翻成 True,另一条不受影响
+    items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
+    assert {it["store_name"]: it["ordered"] for it in items} == {
+        "海底捞(朝阳店)": True,
+        "没下过单的店": False,
+    }
+
+    # 别人的下单不该影响本人标记(_ordered_shop_names 按 user_id 过滤)
+    token_b = _login(client, "13800002011")
+    client.post("/api/v1/compare/record", json=_food_payload("ord-b"), headers=_auth(token_b))
+    items_b = client.get("/api/v1/compare/records", headers=_auth(token_b)).json()["items"]
+    assert [it["ordered"] for it in items_b] == [False]
+
+
+def test_records_list_omits_raw_payload(client) -> None:
+    """列表出参不含 raw_payload(仓库层 defer 掉了重型 JSON 列);详情接口照常返回。
+
+    defer 的列一旦被 ORM 实例读到会触发逐行懒加载(N+1),而列表 schema 本就不该带 raw_payload
+    —— 这条同时守住「列表不泄露上报体全量」和「没人不小心把它加回出参」。
+    """
+    token = _login(client, "13800002012")
+    rid = client.post(
+        "/api/v1/compare/record", json=_food_payload("no-raw"), headers=_auth(token)
+    ).json()["id"]
+
+    items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
+    assert len(items) == 1
+    assert "raw_payload" not in items[0]
+    # 概要字段照常齐全(defer 没误伤列表要用的列)
+    assert items[0]["store_name"] == "海底捞(朝阳店)"
+    assert items[0]["best_platform_id"] == "meituan"
+    assert items[0]["comparison_results"] and items[0]["items"]
+
+    # 详情不 defer:raw_payload 全量还在
+    d = client.get(f"/api/v1/compare/records/{rid}", headers=_auth(token)).json()
+    assert d["raw_payload"]["trace_id"] == "no-raw"
+
+
+def test_records_ordered_filter(client) -> None:
+    """ordered=true 只出「已下单」的记录,且过滤结果自身能翻页。
+
+    这个筛选必须在服务端做:客户端早先是对「已经拉回来的那一页」做 filter,分页之后一页里
+    很可能一条已下单都没有 ——「已下单」tab 就会看着像空的,得手动翻很多页才蹦出一条。
+    """
+    token = _login(client, "13800002013")
+
+    # 3 条「下过单的店」+ 2 条没下过单的店,交错写入,确保过滤不是靠顺序碰巧对上
+    for i in range(3):
+        p = _food_payload(f"of-ordered-{i}")
+        p["store_name"] = "下过单的店"
+        client.post("/api/v1/compare/record", json=p, headers=_auth(token))
+        if i < 2:
+            q = _food_payload(f"of-plain-{i}")
+            q["store_name"] = "没下过单的店"
+            client.post("/api/v1/compare/record", json=q, headers=_auth(token))
+
+    client.post(
+        "/api/v1/order/report",
+        json={
+            "client_event_id": "evt-ordered-filter",
+            "platform": "美团",
+            "platform_package": "com.sankuai.meituan",
+            "pay_channel": "wechat",
+            "compared_price_cents": 12350,
+            "paid_amount_cents": 12350,
+            "shop_name": "下过单的店",
+            "original_price_cents": 12850,
+        },
+        headers=_auth(token),
+    )
+
+    # 不传 ordered:5 条全出(「全部记录」tab 口径不变)
+    assert len(client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]) == 5
+
+    # ordered=true:只出那 3 条,且每条都自带 ordered=True
+    page = client.get("/api/v1/compare/records?ordered=true", headers=_auth(token)).json()
+    assert [it["store_name"] for it in page["items"]] == ["下过单的店"] * 3
+    assert all(it["ordered"] for it in page["items"])
+    assert page["next_cursor"] is None
+
+    # 游标只在「已下单」集合内走 —— 不会把没下单的记录算进一页的 limit 里
+    p1 = client.get(
+        "/api/v1/compare/records?ordered=true&limit=2", headers=_auth(token)
+    ).json()
+    assert len(p1["items"]) == 2
+    assert p1["next_cursor"] is not None
+    p2 = client.get(
+        f"/api/v1/compare/records?ordered=true&limit=2&cursor={p1['next_cursor']}",
+        headers=_auth(token),
+    ).json()
+    assert [it["store_name"] for it in p2["items"]] == ["下过单的店"]
+    # 两页不重叠,合起来正好 3 条
+    assert len({it["id"] for it in p1["items"] + p2["items"]}) == 3
+
+    # 别人的下单不该让本人记录进「已下单」
+    token_b = _login(client, "13800002014")
+    pb = _food_payload("of-b")
+    pb["store_name"] = "下过单的店"
+    client.post("/api/v1/compare/record", json=pb, headers=_auth(token_b))
+    assert client.get(
+        "/api/v1/compare/records?ordered=true", headers=_auth(token_b)
+    ).json()["items"] == []
+
+
+def test_records_keyword_search(client) -> None:
+    """keyword 按店名 / 菜名模糊搜(忽略大小写),LIKE 通配符只当字面量;搜索结果也能翻页。
+
+    菜名走写路径派生的 product_names 文本列 —— items 是 JSON,SQLite 下中文被 ensure_ascii
+    转义,直接 LIKE 搜不到。
+    """
+    token = _login(client, "13800002015")
+
+    b = _food_payload("kw-b")
+    b["store_name"] = "Pizza Hut"
+    b["items"] = [{"name": "榴莲比萨", "qty": 1}]
+    client.post("/api/v1/compare/record", json=b, headers=_auth(token))
+
+    c = _food_payload("kw-c")
+    c["store_name"] = "100%纯牛肉汉堡"
+    c["items"] = [{"name": "双层牛肉堡", "qty": 1}]
+    client.post("/api/v1/compare/record", json=c, headers=_auth(token))
+
+    # 默认 payload 的店名是「海底捞(朝阳店)」
+    client.post("/api/v1/compare/record", json=_food_payload("kw-a"), headers=_auth(token))
+
+    def _search(kw: str, **extra) -> list[str]:
+        r = client.get(
+            "/api/v1/compare/records",
+            params={"keyword": kw, **extra},
+            headers=_auth(token),
+        )
+        assert r.status_code == 200, r.text
+        return [it["store_name"] for it in r.json()["items"]]
+
+    assert _search("海底捞") == ["海底捞(朝阳店)"]          # 店名命中
+    assert _search("榴莲") == ["Pizza Hut"]                  # 菜名命中(product_names)
+    assert _search("pizza") == ["Pizza Hut"]                 # 忽略大小写
+    assert _search("PIZZA") == ["Pizza Hut"]
+    assert _search("不存在的店") == []                        # 没命中就是空
+    # 通配符只当普通字符:搜 % 不该把整表拉回来,搜 _ 也不该匹配任意单字符
+    assert _search("%") == ["100%纯牛肉汉堡"]
+    assert _search("_") == []
+    # 纯空白等同不传 → 不过滤
+    assert len(_search("   ")) == 3
+
+    # 搜索结果自身可翻页
+    for i in range(3):
+        p = _food_payload(f"kw-page-{i}")
+        p["store_name"] = "连锁烤鱼店"
+        client.post("/api/v1/compare/record", json=p, headers=_auth(token))
+    p1 = client.get(
+        "/api/v1/compare/records",
+        params={"keyword": "烤鱼", "limit": 2},
+        headers=_auth(token),
+    ).json()
+    assert len(p1["items"]) == 2
+    assert p1["next_cursor"] is not None
+    p2 = client.get(
+        "/api/v1/compare/records",
+        params={"keyword": "烤鱼", "limit": 2, "cursor": p1["next_cursor"]},
+        headers=_auth(token),
+    ).json()
+    assert [it["store_name"] for it in p2["items"]] == ["连锁烤鱼店"]
+    assert len({it["id"] for it in p1["items"] + p2["items"]}) == 3
+
+
 def test_requires_auth(client) -> None:
     """不带 token 统一 401。"""
     assert client.post("/api/v1/compare/record", json={"trace_id": "t"}).status_code == 401

tests/test_comparison_admin_summary.py

@@ -4,12 +4,28 @@ from __future__ import annotations
 from datetime import UTC, date, datetime
 
 import pytest
+from sqlalchemy.dialects import postgresql
 
 from app.admin.repositories import queries
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 
 
+def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
+    stmt = queries._comparison_duration_aggregate_stmt(
+        [], "success", (0.05, 0.5, 0.95, 0.99)
+    )
+    sql = str(
+        stmt.compile(
+            dialect=postgresql.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert sql.count("percentile_cont") == 4
+    assert "comparison_record.status = 'success'" in sql
+
+
 def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
     db = SessionLocal()
     try:

tests/test_coupon_claim_event.py

@@ -0,0 +1,54 @@
+"""逐次单券事件不能被同设备同日的每日去重记录串场。"""
+
+from sqlalchemy import delete, select
+
+from app.admin.repositories.coupon_data import _point_scores_by_trace, coupon_point_details
+from app.db.session import SessionLocal
+from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord
+from app.repositories.coupon_state import record_claims
+
+
+def test_same_device_same_day_keeps_scores_for_each_trace() -> None:
+    db = SessionLocal()
+    device = "event-repeat-device"
+    first_trace = "event-repeat-first"
+    second_trace = "event-repeat-second"
+    first_results = [
+        {"coupon_id": "mt-repeat", "name": "美团测试券", "status": "success"},
+        {"coupon_id": "tb-repeat", "name": "淘宝测试券", "status": "failed"},
+    ]
+    second_results = [
+        {"coupon_id": "mt-repeat", "name": "美团测试券", "status": "already_claimed"},
+        {"coupon_id": "tb-repeat", "name": "淘宝测试券", "status": "success"},
+    ]
+    try:
+        record_claims(
+            db, device, None, first_trace, first_results, app_env="dev"
+        )
+        record_claims(
+            db, device, None, second_trace, second_results, app_env="prod"
+        )
+
+        assets = db.execute(
+            select(CouponClaimRecord).where(CouponClaimRecord.device_id == device)
+        ).scalars().all()
+        assert len(assets) == 2
+        assert {row.trace_id for row in assets} == {first_trace}
+
+        events = db.execute(
+            select(CouponClaimEvent).where(CouponClaimEvent.device_id == device)
+        ).scalars().all()
+        assert len(events) == 4
+        assert {row.trace_id for row in events} == {first_trace, second_trace}
+
+        scores = _point_scores_by_trace(db, [first_trace, second_trace])
+        assert scores[first_trace] == {"succeeded": 1, "tried": 2}
+        assert scores[second_trace] == {"succeeded": 2, "tried": 2}
+        assert [row["status"] for row in coupon_point_details(
+            db, trace_id=second_trace
+        )] == ["already_claimed", "success"]
+    finally:
+        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.device_id == device))
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == device))
+        db.commit()
+        db.close()

tests/test_coupon_platform_success.py

@@ -7,8 +7,12 @@ from __future__ import annotations
 from datetime import UTC, date, datetime
 
 from sqlalchemy import delete, func, select
+from sqlalchemy.dialects import postgresql
 
-from app.admin.repositories.coupon_data import coupon_data_report
+from app.admin.repositories.coupon_data import (
+    _coupon_summary_aggregate_stmt,
+    coupon_data_report,
+)
 from app.db.session import SessionLocal
 from app.models.coupon_state import CouponSession
 from app.repositories.coupon_state import (
@@ -19,7 +23,12 @@ from app.repositories.coupon_state import (
 
 
 def _agg_session(
-    trace: str, platforms, platform_success, *, status: str = "completed"
+    trace: str,
+    platforms,
+    platform_success,
+    *,
+    status: str = "completed",
+    elapsed_ms: int | None = None,
 ) -> CouponSession:
     """构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
     return CouponSession(
@@ -31,9 +40,53 @@ def _agg_session(
         platform_success=platform_success,
         started_at=datetime(2020, 1, 2, tzinfo=UTC),
         started_date=date(2020, 1, 2),
+        elapsed_ms=elapsed_ms,
     )
 
 
+def test_postgresql_coupon_summary_uses_ordered_set_aggregates() -> None:
+    sql = str(
+        _coupon_summary_aggregate_stmt([]).compile(
+            dialect=postgresql.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert sql.count("percentile_cont") == 4
+    assert "FILTER (WHERE coupon_session.status = 'completed'" in sql
+
+
+def test_coupon_duration_summary_uses_only_completed_rows() -> None:
+    db = SessionLocal()
+    try:
+        db.add_all([
+            _agg_session("duration-a", [], [], elapsed_ms=1000),
+            _agg_session("duration-b", [], [], elapsed_ms=3000),
+            _agg_session(
+                "duration-failed", [], [], status="failed", elapsed_ms=100_000
+            ),
+        ])
+        db.flush()
+
+        summary = coupon_data_report(
+            db,
+            date_from="2020-01-02",
+            date_to="2020-01-02",
+            app_env="prod",
+        )["summary"]
+
+        assert summary["started_count"] == 3
+        assert summary["completed_count"] == 2
+        assert summary["avg_elapsed_ms"] == 2000
+        assert summary["p5_ms"] == 1100
+        assert summary["p50_ms"] == 2000
+        assert summary["p95_ms"] == 2900
+        assert summary["p99_ms"] == 2980
+    finally:
+        db.rollback()
+        db.close()
+
+
 def _make_session(db, trace_id: str, **kw) -> CouponSession:
     row = CouponSession(
         trace_id=trace_id,

tests/test_coupon_point_score.py

@@ -13,7 +13,7 @@ from app.admin.repositories.coupon_data import (
 )
 from app.admin.security import create_admin_token
 from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimEvent, CouponSession
 
 
 def test_point_scores_by_trace() -> None:
@@ -22,14 +22,14 @@ def test_point_scores_by_trace() -> None:
     trace = "point-score-trace"
     try:
         db.add_all([
-            CouponClaimRecord(
+            CouponClaimEvent(
+                trace_id=trace,
                 device_id="score-device",
                 coupon_id=f"mt-score-{status}",
                 claim_date=date(2020, 1, 2),
                 status=status,
                 coupon_name=f"测试点位-{status}",
                 reason="测试失败" if status == "failed" else None,
-                trace_id=trace,
             )
             for status in ("success", "already_claimed", "failed", "skipped")
         ])
@@ -53,12 +53,12 @@ def test_skipped_detail_does_not_create_a_score() -> None:
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
-        db.add(CouponClaimRecord(
+        db.add(CouponClaimEvent(
+            trace_id=trace,
             device_id="score-device-skipped",
             coupon_id="mt-score-skipped-only",
             claim_date=date(2020, 1, 2),
             status="skipped",
-            trace_id=trace,
         ))
         db.flush()
 
@@ -87,12 +87,12 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
             started_date=report_date,
         ))
         db.add_all([
-            CouponClaimRecord(
+            CouponClaimEvent(
+                trace_id=trace,
                 device_id="score-report-device",
                 coupon_id=f"mt-report-{status}",
                 claim_date=report_date,
                 status=status,
-                trace_id=trace,
             )
             for status in ("success", "failed")
         ])
@@ -128,14 +128,14 @@ def test_coupon_point_details_endpoint() -> None:
                 role="super_admin",
             )
         token, _expires_at = create_admin_token(admin_id=admin.id, role=admin.role)
-        db.add(CouponClaimRecord(
+        db.add(CouponClaimEvent(
+            trace_id=trace,
             device_id="point-details-endpoint-device",
             coupon_id="mt-point-details-endpoint",
             coupon_name="接口测试券",
             claim_date=date(2020, 1, 5),
             status="failed",
             reason="接口测试失败",
-            trace_id=trace,
         ))
         db.commit()
 
@@ -156,6 +156,6 @@ def test_coupon_point_details_endpoint() -> None:
         }
     finally:
         db.rollback()
-        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
+        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.trace_id == trace))
         db.commit()
         db.close()

tests/test_coupon_slots.py

@@ -7,7 +7,7 @@ from sqlalchemy import delete, select
 
 from app.admin.repositories.coupon_data import coupon_slot_report
 from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
 from app.repositories.coupon_state import record_claims, session_app_env
 
 
@@ -49,6 +49,7 @@ def test_record_claims_stamps_app_env() -> None:
         assert row.app_env == "prod"
         assert row.status == "already_claimed"
     finally:
+        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.trace_id == "t-stamp"))
         db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == dev))
         db.commit()
         db.close()

tests/test_cps_reconcile_worker.py

@@ -0,0 +1,197 @@
+"""美团、京东 CPS 每日自动对账 worker。"""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime, timedelta
+
+import pytest
+
+from app.core import cps_reconcile_worker as worker
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.integrations.meituan import MeituanCpsError
+
+
+def _configure_platforms(monkeypatch) -> None:
+    monkeypatch.setattr(settings, "MT_CPS_APP_KEY", "mt-key")
+    monkeypatch.setattr(settings, "MT_CPS_APP_SECRET", "mt-secret")
+    monkeypatch.setattr(settings, "JD_UNION_APP_KEY", "jd-key")
+    monkeypatch.setattr(settings, "JD_UNION_APP_SECRET", "jd-secret")
+    monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_LOOKBACK_DAYS", 3)
+
+
+def test_reconcile_once_pulls_meituan_and_jd_by_update_time(monkeypatch) -> None:
+    _configure_platforms(monkeypatch)
+    calls: dict[str, dict] = {}
+
+    def fake_meituan(db, **kwargs):
+        calls["meituan"] = kwargs
+        return {
+            "fetched": 2,
+            "inserted": 1,
+            "updated": 1,
+            "pages": 1,
+            "api_requests": 1,
+        }
+
+    def fake_jd(db, **kwargs):
+        calls["jd"] = kwargs
+        return {
+            "fetched": 3,
+            "inserted": 2,
+            "updated": 1,
+            "pages": 2,
+            "api_requests": 72,
+            "windows": 72,
+        }
+
+    monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fake_meituan)
+    monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
+    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
+    now = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
+
+    result = worker._reconcile_once(now)
+
+    assert result["errors"] == {}
+    assert result["meituan"]["fetched"] == 2
+    assert result["jd"]["fetched"] == 3
+    assert calls["meituan"]["query_time_type"] == 2
+    assert calls["jd"]["query_time_type"] == 3
+    assert calls["meituan"]["end_time"] == int(now.timestamp())
+    assert calls["meituan"]["start_time"] == int((now - timedelta(days=3)).timestamp())
+    assert calls["jd"]["start_time"] == now - timedelta(days=3)
+    assert calls["jd"]["end_time"] == now
+    assert calls["meituan"]["audit_context"]["run_id"] == result["run_id"]
+    assert calls["jd"]["audit_context"]["run_id"] == result["run_id"]
+    assert result["status"] == "success"
+    assert result["manual_retry_required"] is False
+
+
+def test_meituan_failure_does_not_block_jd(monkeypatch) -> None:
+    _configure_platforms(monkeypatch)
+    jd_called = False
+
+    def fail_meituan(db, **kwargs):
+        raise MeituanCpsError("temporary failure")
+
+    def fake_jd(db, **kwargs):
+        nonlocal jd_called
+        jd_called = True
+        return {"fetched": 1, "inserted": 1, "updated": 0, "pages": 1}
+
+    monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fail_meituan)
+    monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
+    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
+
+    result = worker._reconcile_once(datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ))
+
+    assert result["errors"]["meituan"] == "temporary failure"
+    assert jd_called is True
+    assert result["jd"]["fetched"] == 1
+    assert result["meituan"]["status"] == "failed"
+    assert result["status"] == "partial_success"
+    assert result["manual_retry_required"] is True
+
+
+def test_should_run_once_after_five_am() -> None:
+    before = datetime(2026, 7, 22, 4, 59, tzinfo=CN_TZ)
+    at_five = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
+
+    assert worker._should_run(None, before, 5) is False
+    assert worker._should_run(None, at_five, 5) is True
+    assert worker._should_run(date(2026, 7, 22), at_five, 5) is False
+
+
+def test_trigger_and_next_run_are_beijing_time() -> None:
+    before_five = datetime(2026, 7, 22, 4, 30, tzinfo=CN_TZ)
+    after_five = datetime(2026, 7, 22, 6, 0, tzinfo=CN_TZ)
+
+    assert worker._trigger_for_run(before_five, after_five, 5) == "scheduled"
+    assert worker._trigger_for_run(after_five, after_five, 5) == "startup_catchup"
+    assert worker._next_run_at(before_five, 5) == datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
+    assert worker._next_run_at(after_five, 5) == datetime(2026, 7, 23, 5, 0, tzinfo=CN_TZ)
+
+
+def test_reconcile_emits_structured_audit_logs(monkeypatch, caplog) -> None:
+    _configure_platforms(monkeypatch)
+    monkeypatch.setattr(
+        worker.cps_repo,
+        "reconcile_orders",
+        lambda db, **kwargs: {
+            "fetched": 2,
+            "inserted": 1,
+            "updated": 1,
+            "pages": 1,
+            "api_requests": 1,
+        },
+    )
+    monkeypatch.setattr(
+        worker.cps_repo,
+        "reconcile_jd_orders",
+        lambda db, **kwargs: {
+            "fetched": 3,
+            "inserted": 2,
+            "updated": 1,
+            "pages": 1,
+            "api_requests": 72,
+            "windows": 72,
+        },
+    )
+    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
+    monkeypatch.setattr(
+        worker,
+        "_cn_now",
+        lambda: datetime(2026, 7, 22, 5, 1, tzinfo=CN_TZ),
+    )
+
+    with caplog.at_level(logging.INFO, logger=worker.logger.name):
+        result = worker._reconcile_once(
+            datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ),
+            run_id="audit-run-1",
+            trigger="scheduled",
+        )
+
+    records = {record.event: record for record in caplog.records if hasattr(record, "event")}
+    assert records["cps_reconcile.started"].run_id == "audit-run-1"
+    assert records["cps_reconcile.started"].lookback_days == 3
+    assert records["cps_reconcile.started"].scheduled_date == "2026-07-22"
+    assert records["cps_reconcile.started"].hostname
+    assert records["cps_reconcile.completed"].task_status == "success"
+    assert records["cps_reconcile.completed"].manual_retry_required is False
+    assert records["cps_reconcile.completed"].meituan_result["fetched"] == 2
+    assert records["cps_reconcile.completed"].jd_result["api_requests"] == 72
+    assert result["next_run_at"] == "2026-07-23T05:00:00+08:00"
+
+
+def test_jd_request_failure_logs_exact_window(monkeypatch, caplog) -> None:
+    start = datetime(2026, 7, 20, 5, 0, tzinfo=CN_TZ)
+
+    def fail_query(**kwargs):
+        raise RuntimeError("upstream timeout")
+
+    monkeypatch.setattr(worker.cps_repo.jd_union, "query_order_rows", fail_query)
+    with SessionLocal() as db, caplog.at_level(logging.ERROR, logger=worker.logger.name):
+        with pytest.raises(RuntimeError, match="upstream timeout"):
+            worker.cps_repo.reconcile_jd_orders(
+                db,
+                start_time=start,
+                end_time=start + timedelta(hours=2),
+                audit_context={"run_id": "audit-run-2", "trigger": "scheduled"},
+            )
+
+    record = next(
+        item
+        for item in caplog.records
+        if getattr(item, "event", "") == "cps_reconcile.request_failed"
+    )
+    assert record.run_id == "audit-run-2"
+    assert record.platform == "jd"
+    assert record.failed_window_start == "2026-07-20T05:00:00+08:00"
+    assert record.failed_window_end == "2026-07-20T06:00:00+08:00"
+    assert record.failed_page == 1
+
+
+def test_start_worker_respects_auto_switch(monkeypatch) -> None:
+    monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_ENABLED", False)
+    assert worker.start_cps_reconcile_worker() is None