perf(admin): aggregate record duration percentiles in PostgreSQL

.env.example

@@ -113,13 +113,6 @@ 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

@@ -1,26 +0,0 @@
-"""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/coupon_claim_event.py

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

app/admin/repositories/coupon_data.py

@@ -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 CouponClaimEvent, CouponClaimRecord, CouponSession
+from app.models.coupon_state import 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
@@ -193,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((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
+    succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
     rows = db.execute(
         select(
-            CouponClaimEvent.trace_id,
+            CouponClaimRecord.trace_id,
             succeeded.label("succeeded"),
             func.count().label("tried"),
         )
         .where(
-            CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimEvent.status.in_(_SLOT_TRIED),
+            CouponClaimRecord.trace_id.in_(trace_ids),
+            CouponClaimRecord.status.in_(_SLOT_TRIED),
         )
-        .group_by(CouponClaimEvent.trace_id)
+        .group_by(CouponClaimRecord.trace_id)
     ).all()
     return {
         trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
@@ -217,13 +217,13 @@ def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
     """按单个 trace 查询逐券结果；仅在后台用户点击分数时调用。"""
     rows = db.execute(
         select(
-            CouponClaimEvent.coupon_id,
-            CouponClaimEvent.coupon_name,
-            CouponClaimEvent.status,
-            CouponClaimEvent.reason,
+            CouponClaimRecord.coupon_id,
+            CouponClaimRecord.coupon_name,
+            CouponClaimRecord.status,
+            CouponClaimRecord.reason,
         )
-        .where(CouponClaimEvent.trace_id == trace_id)
-        .order_by(CouponClaimEvent.id)
+        .where(CouponClaimRecord.trace_id == trace_id)
+        .order_by(CouponClaimRecord.id)
     ).all()
     return [
         {

app/admin/repositories/cps.py

@@ -5,7 +5,6 @@
 """
 from __future__ import annotations
 
-import logging
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from typing import Any
@@ -16,14 +15,12 @@ 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"}
@@ -381,35 +378,18 @@ 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 = api_requests = 0
+    fetched = inserted = updated = pages = 0
     page = 1
     while page <= max_pages:
-        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
+        resp = meituan.query_order(
+            sid=sid, start_time=start_time, end_time=end_time,
+            query_time_type=query_time_type, page=page, limit=100,
+        )
         rows = ((resp.get("data") or {}).get("dataList")) or []
         if not rows:
             break
@@ -439,58 +419,30 @@ def reconcile_orders(
             break
         page += 1
     db.commit()
-    return {
-        "fetched": fetched,
-        "inserted": inserted,
-        "updated": updated,
-        "pages": pages,
-        "api_requests": api_requests,
-    }
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
 
 
 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 = api_requests = windows = 0
+    fetched = inserted = updated = pages = 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:
-            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
+            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,
+            )
             rows = resp.get("rows") or []
             has_more = bool(resp.get("has_more"))
             if not rows:
@@ -517,14 +469,7 @@ def reconcile_jd_orders(
             page += 1
         cur = win_end
     db.commit()
-    return {
-        "fetched": fetched,
-        "inserted": inserted,
-        "updated": updated,
-        "pages": pages,
-        "api_requests": api_requests,
-        "windows": windows,
-    }
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
 
 
 def list_orders(

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/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 环境，给每日资产和逐次事件同时打环境标。
+        # 取本次 session 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)。
         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()
 
-    # 领券结果沉淀:每日资产 + 逐次事件；中间帧和 done 全量帧均幂等写库。
+    # 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
     # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
     if device_id:
         results = _extract_coupon_results(resp_json)

app/core/config.py

@@ -188,13 +188,6 @@ 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

@@ -1,428 +0,0 @@
-"""美团、京东 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,10 +43,6 @@ 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,
@@ -93,7 +89,6 @@ 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()
@@ -103,7 +98,6 @@ 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,7 +21,6 @@ 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/coupon_state.py

@@ -96,40 +96,6 @@ 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/coupon_state.py

@@ -14,7 +14,6 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.models.coupon_state import (
-    CouponClaimEvent,
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
@@ -165,12 +164,11 @@ 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)。
-    - CouponClaimRecord 按 (device, coupon_id, 今天) 幂等，供每日资产口径使用。
-    - CouponClaimEvent 按 (trace_id, coupon_id) 幂等，供 admin 逐场统计使用。
+    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
     """
     today = today_cn()
     written = 0
@@ -210,41 +208,6 @@ 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

scripts/seed_coupon_session_mock.py

@@ -1,197 +0,0 @@
-"""生成 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_coupon_claim_event.py

@@ -1,54 +0,0 @@
-"""逐次单券事件不能被同设备同日的每日去重记录串场。"""
-
-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_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 CouponClaimEvent, CouponSession
+from app.models.coupon_state import CouponClaimRecord, 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([
-            CouponClaimEvent(
-                trace_id=trace,
+            CouponClaimRecord(
                 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(CouponClaimEvent(
-            trace_id=trace,
+        db.add(CouponClaimRecord(
             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([
-            CouponClaimEvent(
-                trace_id=trace,
+            CouponClaimRecord(
                 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(CouponClaimEvent(
-            trace_id=trace,
+        db.add(CouponClaimRecord(
             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(CouponClaimEvent).where(CouponClaimEvent.trace_id == trace))
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.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 CouponClaimEvent, CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 from app.repositories.coupon_state import record_claims, session_app_env
 
 
@@ -49,7 +49,6 @@ 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

@@ -1,197 +0,0 @@
-"""美团、京东 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