feat(ad-ecpm): trace_id 端到端落库 — 模型/迁移/Schema/Repository/Endpoint(A1-A3+A5)

- AdEcpmRecord 新增 trace_id(String 64, index, nullable)列,与 ad_feed_reward_record 一致
- 迁移 ad_ecpm_trace_id:ADD COLUMN + CREATE INDEX,同时收敛双 head
  (11c44afbea58 selfstat + merge_pages_override_coupon_slot → 单 head)
- EcpmReportIn.trace_id 字段透传客户端本场比价/领券 session ID
- create_ecpm_record 新增 trace_id 参数并落库
- ecpm_report 端点 forward payload.trace_id 到 repository

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

alembic/versions/ad_ecpm_trace_id.py

@@ -0,0 +1,45 @@
+"""ad_ecpm_record.trace_id(展示收益归属到比价/领券 trace)+ 收敛双 head
+
+信息流(Draw)展示 eCPM 上报时带上本场比价/领券 trace_id,落此列;领券数据 / 比价记录看板
+按 trace_id 聚合"本次广告收益"。激励视频/福利/旧客户端为 NULL。
+
+顺带把当前两个 head(11c44afbea58 selfstat 表 + merge_pages_override_coupon_slot)收敛成
+单 head,让 `alembic upgrade head`(单数,部署/run.sh 用)恢复正常。
+
+Revision ID: ad_ecpm_trace_id
+Revises: 11c44afbea58, merge_pages_override_coupon_slot
+Create Date: 2026-07-10
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "ad_ecpm_trace_id"
+down_revision: Union[str, Sequence[str], None] = (
+    "11c44afbea58",
+    "merge_pages_override_coupon_slot",
+)
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # SQLite 下 ADD COLUMN(可空)与 CREATE INDEX 均原生支持,无需 batch_alter_table
+    # (同 ad_feed_reward_trace_id 迁移)。
+    op.add_column(
+        "ad_ecpm_record",
+        sa.Column("trace_id", sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_ecpm_record_trace_id"),
+        "ad_ecpm_record",
+        ["trace_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_ad_ecpm_record_trace_id"), table_name="ad_ecpm_record")
+    op.drop_column("ad_ecpm_record", "trace_id")

app/api/v1/ad.py

@@ -286,6 +286,7 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
         ad_session_id=payload.ad_session_id,
         adn=payload.adn, slot_id=payload.slot_id,
         feed_scene=payload.feed_scene,
+        trace_id=payload.trace_id,
         app_env=payload.app_env, our_code_id=payload.our_code_id,
     )
     logger.info(

app/models/ad_ecpm.py

@@ -32,6 +32,9 @@ class AdEcpmRecord(Base):
     # 点位场景:comparison(比价) / coupon(领券) / welfare(福利),供收益报表区分比价/领券 Draw 收益;
     # 仅信息流/Draw 上报(比价与领券共用同一代码位,只能客户端各调用点显式打标),激励视频为 NULL。
     feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    # 本次比价/领券 trace_id(信息流场景客户端带上):把这条展示收益归属到对应比价/领券记录。
+    # 领券数据 / 比价记录看板按 trace_id 聚合"本次广告收益"。激励视频/福利/旧客户端 = NULL。
+    trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     # 客户端生成的一次广告会话 id;激励视频 S2S 回调 extra 会透传同值
     ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     # 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)

app/repositories/ad_ecpm.py

@@ -10,6 +10,7 @@ from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.core import rewards
 from app.core.rewards import cn_today
 from app.models.ad_ecpm import AdEcpmRecord
 
@@ -24,6 +25,7 @@ def create_ecpm_record(
     adn: str | None = None,
     slot_id: str | None = None,
     feed_scene: str | None = None,
+    trace_id: str | None = None,
     app_env: str | None = None,
     our_code_id: str | None = None,
 ) -> AdEcpmRecord:
@@ -43,6 +45,7 @@ def create_ecpm_record(
         adn=adn,
         slot_id=slot_id,
         feed_scene=feed_scene,
+        trace_id=trace_id,
         app_env=app_env,
         our_code_id=our_code_id,
         ecpm_raw=ecpm_raw,
@@ -105,3 +108,27 @@ def count_today(db: Session, user_id: int) -> int:
             AdEcpmRecord.report_date == cn_today().isoformat(),
         )
     ).scalar_one()
+
+
+def revenue_yuan_by_trace(db: Session, trace_ids: list[str]) -> dict[str, float]:
+    """各 trace_id 的广告预估收益(元):按 trace_id 聚合 ad_ecpm_record 的展示收益。
+
+    单条展示收益 = min(eCPM元, AD_ECPM_MAX_FEN/100) / 1000(与 admin 广告收益报表同口径)。
+    ecpm_raw 是字符串且需逐条钳顶,故取回后 Python 求和(行数=本页各 trace 的展示条数,很小)。
+    trace_id 仅信息流(比价/领券)场景客户端带,激励视频/旧数据为 NULL,按 trace_id 过滤天然只算对应场景。
+    只喂**当前页**的 trace_id(≤ 一页条数);空集合直接返回(避免 IN () 非法)。
+    """
+    if not trace_ids:
+        return {}
+    rows = db.execute(
+        select(AdEcpmRecord.trace_id, AdEcpmRecord.ecpm_raw).where(
+            AdEcpmRecord.trace_id.in_(trace_ids),
+        )
+    ).all()
+    cap_yuan = rewards.AD_ECPM_MAX_FEN / 100.0
+    out: dict[str, float] = {}
+    for tid, ecpm_raw in rows:
+        if not tid:
+            continue
+        out[tid] = out.get(tid, 0.0) + min(rewards.parse_ecpm_yuan(ecpm_raw), cap_yuan) / 1000.0
+    return {tid: round(v, 6) for tid, v in out.items()}

app/schemas/ad.py

@@ -63,6 +63,12 @@ class EcpmReportIn(BaseModel):
         description="点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页);"
         "比价与领券共用同一 Draw 代码位,需客户端在各调用点显式标注,供收益报表区分比价/领券;激励视频为空",
     )
+    trace_id: str | None = Field(
+        None,
+        max_length=64,
+        description="本次比价/领券 trace_id(信息流场景带上):把这条展示收益归属到对应比价/领券,"
+        "供领券数据/比价记录看板聚合本场广告收益;激励视频/福利为空",
+    )
     app_env: str | None = Field(
         None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)"
     )