perf(pricebot): 透传复用共享 httpx 单例,免每请求重建 SSL 上下文 + 绕过进程代理

coupon/compare 透传原先每请求 async with httpx.AsyncClient(...) 新建:
每次重建 SSL 上下文(加载 certifi CA)实测 ~1s+,而 pricebot 是纯 HTTP
透传根本用不到 TLS;且 trust_env 默认 True 会把 http://localhost:8000
经进程代理(Clash)再绕几秒。

抽 app/core/pricebot_client.py 共享单例:trust_env=False 直连,lifespan
启动预热(SSL 一次性成本付在启动)、关停 aclose;timeout 下放到 .post()
保留 coupon 30s / compare 60s 差异。keep-alive 复用 TCP,每帧降到个位数 ms。

app/admin/repositories/ad_audit.py

@@ -4,9 +4,9 @@
 - 看视频:每条 granted = 1 份,第 N 份 = 该用户 granted 的 reward_video **账号累计**顺序号
   (与 ad_reward.grant_ad_reward 里 `_granted_cumulative + 1` 一致;LT 因子不按天重置,
   故复算时要把当日序号叠加上该用户在本日**之前**的累计已发份数)。
-- 信息流:每条按 unit_count 份逐份累加,LT 序号 = 该用户 granted 份数**账号累计**
+- 信息流:**每条 granted = 1 份**(与 ad_feed_reward.grant_feed_reward 同口径:看满一份即发该条
-  (与 ad_feed_reward._unit_reward_total 的 existing_units 一致;同样不按天重置,
+  满额,**不按 unit_count 逐份累加**),LT 序号 = 该用户 granted **条数**账号累计
-  复算需叠加本日之前的累计份数)。
+  (与 ad_feed_reward.granted_unit_total 的 COUNT 一致;不按天重置,复算需叠加本日之前的累计条数)。
 
 非 granted(capped/ecpm_missing)不占用份序号、应发恒 0,据此校验闸口是否确实没发。
 """
@@ -108,14 +108,18 @@ def _reward_video_rows(
     return rows
 
 
-def _feed_prior_granted_units(
+def _feed_prior_granted_count(
     db: Session, *, date: str, user_id: int | None
 ) -> dict[int, int]:
-    """各用户在 date **之前** granted 的信息流份数累计,作为当日复算的 LT 序号起点。"""
+    """各用户在 date **之前** granted 的信息流**条数**累计,作为当日复算的 LT 序号起点。
+
+    与发奖侧 ad_feed_reward.granted_unit_total(COUNT status=granted)对齐:一条广告 = 1 份,
+    LT 按账号累计**条数**递进。**不再用 SUM(unit_count)**——那是「一条按时长折多份」的过时口径,
+    与现行发奖(每条 1 份)漂移,会让 unit_count>1 的记录复算虚高、对账恒「不符」。"""
     stmt = (
         select(
             AdFeedRewardRecord.user_id,
-            func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0),
+            func.count(),
         )
         .where(
             AdFeedRewardRecord.reward_date < date,
@@ -144,10 +148,11 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
 def _feed_rows(
     db: Session, *, date: str, user_id: int | None, scene: str | None = None
 ) -> list[dict]:
-    """信息流记录复算。granted 记录逐份累加,LT 序号沿用账号累计份数(含本日之前)。
+    """信息流记录复算。**每条 granted = 1 份**(与发奖同口径,不按 unit_count 累加),
+    LT 序号沿用账号累计**条数**(含本日之前)。
 
-    **关键:LT 因子账号累计按全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
+    **关键:LT 因子账号累计按全表 granted 条数累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
-    故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_units 累加;scene 只决定
+    故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_count 累加;scene 只决定
     哪些行被**留下展示**(由 _feed_scene_matches 判断),不影响累计基线,保证复算序号与正式发奖一致。
     """
     stmt = (
@@ -158,23 +163,20 @@ def _feed_rows(
     if user_id is not None:
         stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
 
-    # 本日之前的累计份数做起点,与 _unit_reward_total 的 existing_units(累计)对齐
+    # 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
-    granted_units: dict[int, int] = _feed_prior_granted_units(db, date=date, user_id=user_id)
+    granted_count: dict[int, int] = _feed_prior_granted_count(db, date=date, user_id=user_id)
     rows: list[dict] = []
     for rec in db.execute(stmt).scalars():
         keep = _feed_scene_matches(rec, scene)  # 累计照常推进,这里只决定是否展示本行
         if rec.status == "granted":
-            existing = granted_units.get(rec.user_id, 0)
+            # 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
-            units = rec.unit_count
+            # nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
-            granted_units[rec.user_id] = existing + units
+            # (即便 scene 不匹配不展示也要 +1,保证序号与正式发奖一致)。
+            nth = granted_count.get(rec.user_id, 0) + 1
+            granted_count[rec.user_id] = nth
             if not keep:
                 continue
-            expected = sum(
+            expected = rewards.calculate_ad_reward_coin(rec.ecpm_raw, nth)
-                rewards.calculate_ad_reward_coin(rec.ecpm_raw, existing + offset)
-                for offset in range(1, units + 1)
-            )
-            start = existing + 1 if units > 0 else None
-            end = existing + units if units > 0 else None
             rows.append({
                 "scene": "feed",
                 "ad_type": rec.ad_type or "feed",
@@ -188,11 +190,11 @@ def _feed_rows(
                 "status": rec.status,
                 "ecpm": rec.ecpm_raw,
                 "ecpm_factor": rewards.ad_ecpm_factor(rewards.parse_ecpm_yuan(rec.ecpm_raw)),
-                "units": units,
+                "units": 1,
-                "lt_index_start": start,
+                "lt_index_start": nth,
-                "lt_index_end": end,
+                "lt_index_end": nth,
-                "lt_factor_start": rewards.ad_lt_factor(start) if start else None,
+                "lt_factor_start": rewards.ad_lt_factor(nth),
-                "lt_factor_end": rewards.ad_lt_factor(end) if end else None,
+                "lt_factor_end": rewards.ad_lt_factor(nth),
                 "expected_coin": expected,
                 "actual_coin": rec.coin,
                 "matched": expected == rec.coin,

app/admin/repositories/ad_revenue.py

@@ -27,6 +27,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.admin.repositories import ad_audit
+from app.admin.repositories import stats as admin_stats
 from app.core import rewards
 from app.models.ad_ecpm import AdEcpmRecord
 from app.models.user import User
@@ -84,14 +85,20 @@ def ad_revenue_report(
     date_to: str,
     user_id: int | None = None,
     ad_type: str | None = None,
+    feed_scene: str | None = None,
     granularity: str = "day",
     limit: int = 500,
+    offset: int = 0,
+    sort: str = "time",
 ) -> dict:
     """日期区间(北京时间,闭区间)**逐条广告事件**列表 + 发奖对账。单日时 date_from==date_to。
 
     每个 item = 一次广告事件(展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行)。
-    ad_type: None=全部 / reward_video / feed / draw。granularity=hour 时每行带北京小时(由各自时间算)。
+    ad_type: None=全部 / reward_video / feed / draw。feed_scene: None=全部 /
-    limit 只截断 items(事件明细),total 与 total_* / daily 在全量上统计,数字始终可信。
+    comparison / coupon / welfare,作为全局筛选(同时作用于明细、合计与 daily/hourly 趋势)。
+    granularity=hour 时每行带北京小时(由各自时间算),并额外返回全量 hourly 序列。
+    事件按时间倒序(新→旧)排列;limit/offset 对排序后的全量做分页切片(items 为当前页),
+    total 与 total_* / daily / hourly 在全量上统计,不受分页影响。
     """
     by_hour = granularity == "hour"
 
@@ -200,7 +207,17 @@ def ad_revenue_report(
             "reward_detail": _reward_detail(row),
         })
 
-    events.sort(key=lambda e: (e["report_date"], e["user_id"], e["created_at"]))
+    # 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
+    # feed_scene 仅信息流 / Draw 有值,激励视频与旧数据为 None;选中后只保留该场景事件。
+    if feed_scene is not None:
+        events = [e for e in events if e.get("feed_scene") == feed_scene]
+
+    # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
+    # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
+    if sort == "ecpm":
+        events.sort(key=lambda e: rewards.parse_ecpm_fen(e["ecpm"]), reverse=True)
+    else:
+        events.sort(key=lambda e: (e["report_date"], e["created_at"]), reverse=True)
 
     # 补手机号(admin 展示用,完整不脱敏,与用户 / 钱包 / 比价记录页一致):批量一次查,避免 N+1。
     uids = {e["user_id"] for e in events}
@@ -238,14 +255,60 @@ def ad_revenue_report(
         for d in sorted(daily_map.values(), key=lambda x: x["date"])
     ]
 
+    # 按小时汇总(全量,不受分页 limit/offset 影响):供前端按小时趋势图(单日 granularity=hour 时用)。
+    # 只在 by_hour 下聚合(此时每个 event 带 hour);否则空。前端按天趋势仍用 daily。
+    hourly: list[dict] = []
+    if by_hour:
+        hour_map: dict[int, dict] = {}
+        for e in events:
+            h = e["hour"]
+            if h is None:
+                continue
+            hd = hour_map.get(h)
+            if hd is None:
+                hd = {"hour": h, "impressions": 0, "revenue_yuan": 0.0,
+                      "expected_coin": 0, "actual_coin": 0}
+                hour_map[h] = hd
+            hd["impressions"] += e["impressions"]
+            hd["revenue_yuan"] += e["revenue_yuan"]
+            hd["expected_coin"] += e["expected_coin"]
+            hd["actual_coin"] += e["actual_coin"]
+        hourly = [
+            {**hd, "revenue_yuan": round(hd["revenue_yuan"], 6)}
+            for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
+        ]
+
+    # 分广告类型小计(按 ad_type:展示条数 + 预估收益;eCPM 由前端用 收益÷展示×1000 算)。
+    # 基于全量(已按 feed_scene 过滤)events;前端只取 draw / reward_video 两类展示。
+    type_map: dict[str, dict] = {}
+    for e in events:
+        t = type_map.get(e["ad_type"])
+        if t is None:
+            t = {"impressions": 0, "revenue_yuan": 0.0}
+            type_map[e["ad_type"]] = t
+        t["impressions"] += e["impressions"]
+        t["revenue_yuan"] += e["revenue_yuan"]
+    type_stats = {
+        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
+        for k, v in type_map.items()
+    }
+
+    # DAU:复用大盘「今日活跃」口径(stats.today_dau,last_login_at)。该口径只能算今日,
+    # 故仅当查询=今日单天时给值;历史 / 多天区间返回 None,前端显示「-」。
+    is_today = date_from == date_to == rewards.cn_today().isoformat()
+    dau = admin_stats.today_dau(db) if is_today else None
+
     return {
         "total": len(events),
-        "truncated": len(events) > limit,
+        "truncated": len(events) > offset + limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
         "total_expected_coin": total_expected_coin,
         "total_actual_coin": total_actual_coin,
         "mismatch_count": mismatch_count,
         "daily": daily,
-        "items": events[:limit],
+        "hourly": hourly,
+        "type_stats": type_stats,
+        "dau": dau,
+        "items": events[offset:offset + limit],
     }

app/admin/repositories/stats.py

@@ -28,6 +28,19 @@ def _beijing_today_start_utc() -> datetime:
     return start_bj.astimezone(timezone.utc)
 
 
+def today_dau(db: Session) -> int:
+    """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。
+
+    大盘与广告收益报表共用此口径,单一来源避免漂移。
+    ⚠️ last_login_at 是单值字段(只存最后一次登录时刻),故只能算「今日」,
+    无法回溯历史某天的 DAU——调用方按此约束决定历史区间是否展示。
+    """
+    today_start = _beijing_today_start_utc()
+    return db.execute(
+        select(func.count(User.id)).where(User.last_login_at >= today_start)
+    ).scalar_one()
+
+
 def dashboard_overview(db: Session) -> dict:
     today_start = _beijing_today_start_utc()
 
@@ -69,7 +82,7 @@ def dashboard_overview(db: Session) -> dict:
             "disabled": by_status.get("disabled", 0),
             "deleted": by_status.get("deleted", 0),
             "new_today": _count(User, User.created_at >= today_start),
-            "dau": _count(User, User.last_login_at >= today_start),
+            "dau": today_dau(db),
         },
         "coins": {
             # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)

app/admin/routers/ad_revenue.py

@@ -11,7 +11,13 @@ from fastapi import APIRouter, Depends, HTTPException, Query
 
 from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import ad_revenue
-from app.admin.schemas.ad_revenue import AdRevenueDaily, AdRevenueReportOut, AdRevenueRow
+from app.admin.schemas.ad_revenue import (
+    AdRevenueDaily,
+    AdRevenueHourly,
+    AdRevenueReportOut,
+    AdRevenueRow,
+    AdRevenueTypeStat,
+)
 from app.core.rewards import cn_today
 
 router = APIRouter(
@@ -43,10 +49,21 @@ def get_ad_revenue_report(
         str | None,
         Query(description="reward_video / feed / draw;不传=全部类型"),
     ] = None,
+    feed_scene: Annotated[
+        str | None,
+        Query(
+            description="comparison(比价) / coupon(领券) / welfare(福利);不传=全部场景。"
+            "全局筛选,同时影响明细 / 合计 / 趋势"
+        ),
+    ] = None,
     granularity: Annotated[
         str, Query(description="day=按天 / hour=按小时(北京时间);区间>1 天建议用 day")
     ] = "day",
-    limit: Annotated[int, Query(ge=1, le=1000)] = 500,
+    limit: Annotated[int, Query(ge=1, le=1000, description="每页条数(分页大小)")] = 500,
+    offset: Annotated[int, Query(ge=0, description="分页偏移(已跳过的条数)=(页码-1)×每页条数")] = 0,
+    sort: Annotated[
+        str, Query(description="排序:time=时间倒序(默认) / ecpm=按 eCPM 数值倒序")
+    ] = "time",
 ) -> AdRevenueReportOut:
     today = cn_today()
     d_from = _parse_day(date_from, field="date_from", default=today)
@@ -58,12 +75,16 @@ def get_ad_revenue_report(
 
     result = ad_revenue.ad_revenue_report(
         db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
-        user_id=user_id, ad_type=ad_type, granularity=granularity, limit=limit,
+        user_id=user_id, ad_type=ad_type, feed_scene=feed_scene,
+        granularity=granularity, limit=limit, offset=offset, sort=sort,
     )
     return AdRevenueReportOut(
         date_from=d_from.isoformat(),
         date_to=d_to.isoformat(),
         daily=[AdRevenueDaily(**d) for d in result["daily"]],
+        hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
+        type_stats={k: AdRevenueTypeStat(**v) for k, v in result["type_stats"].items()},
+        dau=result["dau"],
         total=result["total"],
         truncated=result["truncated"],
         total_impressions=result["total_impressions"],

app/admin/schemas/ad_revenue.py

@@ -40,7 +40,7 @@ class AdRevenueRecord(BaseModel):
 
 
 class AdRevenueDaily(BaseModel):
-    """按日期汇总的一天(供前端按天趋势图;全量,不受 limit 影响)。"""
+    """按日期汇总的一天(供前端按天趋势图;全量,不受分页影响)。"""
 
     date: str = Field(..., description="北京时间 YYYY-MM-DD")
     impressions: int = Field(..., description="当天展示条数合计")
@@ -49,6 +49,23 @@ class AdRevenueDaily(BaseModel):
     actual_coin: int = Field(..., description="当天实发金币合计")
 
 
+class AdRevenueHourly(BaseModel):
+    """按北京小时(0–23)汇总的一小时(供前端按小时趋势图;全量,不受分页影响,单日 granularity=hour 时非空)。"""
+
+    hour: int = Field(..., description="北京时间小时 0–23")
+    impressions: int = Field(..., description="该小时展示条数合计")
+    revenue_yuan: float = Field(..., description="该小时预估收益合计(元)")
+    expected_coin: int = Field(..., description="该小时应发金币合计")
+    actual_coin: int = Field(..., description="该小时实发金币合计")
+
+
+class AdRevenueTypeStat(BaseModel):
+    """按广告类型(ad_type)的小计:展示条数 + 预估收益(eCPM 由前端用 收益÷展示×1000 算)。"""
+
+    impressions: int = Field(..., description="该类型展示条数合计")
+    revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
+
+
 class AdRevenueRow(BaseModel):
     """一次广告事件(逐条一行):激励视频展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行。"""
 
@@ -91,8 +108,20 @@ class AdRevenueReportOut(BaseModel):
     date_from: str = Field(..., description="报表起始日期(北京时间 YYYY-MM-DD)")
     date_to: str = Field(..., description="报表结束日期(北京时间 YYYY-MM-DD,闭区间;单日时与 date_from 相同)")
     daily: list[AdRevenueDaily] = Field(..., description="按日期汇总序列(全量,供按天趋势图)")
-    total: int = Field(..., description="广告事件总数(全量,不受 limit 影响)")
+    hourly: list[AdRevenueHourly] = Field(
-    truncated: bool = Field(..., description="明细是否被 limit 截断")
+        default_factory=list,
+        description="按小时汇总序列(全量,供按小时趋势图;按天查询时为空)",
+    )
+    type_stats: dict[str, AdRevenueTypeStat] = Field(
+        default_factory=dict,
+        description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
+    )
+    dau: int | None = Field(
+        None,
+        description="今日活跃用户数(复用大盘口径,last_login_at);**仅查询=今日单天时有值**,历史/多天为 null",
+    )
+    total: int = Field(..., description="广告事件总数(全量,不受分页影响;= 当前筛选下的分页总条数)")
+    truncated: bool = Field(..., description="当前页之后是否还有更多事件(len(events) > offset + limit)")
     total_impressions: int = Field(..., description="全量展示条数合计")
     total_revenue_yuan: float = Field(..., description="全量收益合计(元)")
     total_expected_coin: int = Field(..., description="全量应发金币合计")

app/api/v1/ad.py

@@ -411,6 +411,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         app_env=payload.app_env,
         our_code_id=payload.our_code_id,
         aborted=payload.aborted,
+        display_coin=payload.display_coin,
     )
     logger.info(
         "feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",

app/api/v1/compare.py

@@ -26,6 +26,7 @@ import httpx
 from fastapi import APIRouter, HTTPException, Request, status
 
 from app.core.config import settings
+from app.core.pricebot_client import get_pricebot_client
 from app.core.pricebot_router import pick_pricebot
 
 logger = logging.getLogger("shagua.compare")
@@ -65,10 +66,10 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
     )
 
     try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
+        client = get_pricebot_client()
-            resp = await client.post(
+        resp = await client.post(
-                url, content=raw, headers={"Content-Type": "application/json"}
+            url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
-            )
+        )
     except httpx.RequestError as e:
         logger.error("[pricebot] request failed: %s", e)
         raise HTTPException(

app/api/v1/coupon.py

@@ -20,6 +20,7 @@ from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import CurrentUser, DbSession
 from app.core.config import settings
+from app.core.pricebot_client import get_pricebot_client
 from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import coupon_state as coupon_repo
@@ -141,10 +142,10 @@ async def coupon_step(
     )
 
     try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
+        client = get_pricebot_client()
-            resp = await client.post(
+        resp = await client.post(
-                url, content=raw, headers={"Content-Type": "application/json"}
+            url, content=raw, headers={"Content-Type": "application/json"}, timeout=timeout
-            )
+        )
     except httpx.RequestError as e:
         logger.error("[pricebot] request failed: %s", e)
         raise HTTPException(

app/core/pricebot_client.py

@@ -0,0 +1,35 @@
+"""透传到 pricebot 的共享 httpx.AsyncClient 单例。
+
+为什么不能每请求新建(coupon.py / compare.py 老写法 async with httpx.AsyncClient(...)):
+  ① 每次构造都重建一套 SSL 上下文(httpx.create_ssl_context 加载 certifi CA),实测
+     ~1s+/次;而 pricebot 是纯 http 透传,根本用不到 TLS → 纯浪费,且每帧重交一次。
+  ② trust_env 默认 True 会读进程 HTTP_PROXY,把 http://localhost:8000 这条本地透传整个
+     塞进本机代理(如 Clash 7897),恒定再多几秒。
+单例:启动只建一次(SSL/连接池一次性),keep-alive 复用 TCP,每帧降到个位数 ms。
+trust_env=False:对齐 integrations/meituan.py 的既有约定,不被进程代理误导,直连 pricebot。
+"""
+from __future__ import annotations
+
+import httpx
+
+_client: httpx.AsyncClient | None = None
+
+
+def get_pricebot_client() -> httpx.AsyncClient:
+    """取透传单例。lifespan 启动会预热;未预热(如测试态)懒建兜底。
+
+    超时不在此固化(coupon 30s / compare 60s 不同),由调用点 client.post(timeout=...) 传。
+    懒建无 await,asyncio 单线程下不会有并发竞态。
+    """
+    global _client
+    if _client is None:
+        _client = httpx.AsyncClient(trust_env=False)
+    return _client
+
+
+async def aclose_pricebot_client() -> None:
+    """lifespan 关停时调,优雅关连接池。"""
+    global _client
+    if _client is not None:
+        await _client.aclose()
+        _client = None

app/main.py

@@ -49,6 +49,7 @@ from app.core.daily_exchange_worker import (
     stop_daily_exchange_worker,
 )
 from app.core.logging import setup_logging
+from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client
 from app.core.withdraw_reconcile_worker import (
     start_withdraw_reconcile_worker,
     stop_withdraw_reconcile_worker,
@@ -68,6 +69,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.APP_DEBUG,
         settings.DATABASE_URL.split("://", 1)[0],
     )
+    get_pricebot_client()  # 预热透传 client:把建 SSL 上下文的一次性成本付在启动,首个领券请求即热
     reconcile_task = start_withdraw_reconcile_worker()
     heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
@@ -77,6 +79,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
         await stop_daily_exchange_worker(daily_exchange_task)
+        await aclose_pricebot_client()
         logger.info("shutting down")
 
 

app/repositories/ad_ecpm.py

@@ -53,11 +53,16 @@ def create_ecpm_record(
         db.commit()
     except IntegrityError:
         db.rollback()
-        if ad_session_id:
+        # 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
-            existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+        # 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
-            if existing is not None:
+        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
-                return existing
+        # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
-        raise
+        # 旧逻辑在此 raise 成 500(本应静默吞掉)。
+        existing = _find_by_session_global(db, ad_session_id)
+        if existing is not None:
+            return existing
+        # 极少:rollback 后既存记录又查不到(并发删除 / 竞态)。吞掉、返回未入库的内存对象(调用方不读返回值)。
+        return rec
     db.refresh(rec)
     return rec
 
@@ -76,6 +81,20 @@ def find_by_session(
     ).scalar_one_or_none()
 
 
+def _find_by_session_global(db: Session, ad_session_id: str | None) -> AdEcpmRecord | None:
+    """按 ad_session_id **全局**查找(与唯一约束 uq_ad_ecpm_record_session 同口径,不含 user_id)。
+
+    仅 create_ecpm_record 撞约束后兜底用:此时撞的是全局会话约束,既存记录可能属于**另一个 user**,
+    带 user_id 的 find_by_session 会漏掉它、导致误判「查无 → raise 500」。其它业务查「某 user 的某次
+    展示 eCPM」仍用 find_by_session(带 user_id,语义更准),不走这里。
+    """
+    if not ad_session_id:
+        return None
+    return db.execute(
+        select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == ad_session_id)
+    ).scalar_one_or_none()
+
+
 def count_today(db: Session, user_id: int) -> int:
     """该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
     return db.execute(

app/repositories/ad_feed_reward.py

@@ -73,16 +73,19 @@ def grant_feed_reward(
     app_env: str | None = None,
     our_code_id: str | None = None,
     aborted: bool = False,
+    display_coin: int = 0,
 ) -> AdFeedRewardRecord:
     """**每条**信息流广告(客户端每条各上报一次)结算奖励。client_event_id 幂等,同号重试不重复发。
 
-    发奖规则:**一条广告 = 一个单次公式值**(rewards.calculate_ad_reward_coin),因子2(LT)按账号累计
+    发奖规则(所见即所得, 2026-06-27 用户拍板「显示多少给多少」):优先**直接发客户端小球显示的金币
-    **条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。
+    display_coin**;防刷钳到本条「1 份满额」(eCPM 已钳 AD_ECPM_MAX_FEN, 因子2 按账号累计已发条数取档),
+    合法显示(实际因子2 × 进度 p ≤ 1 份)不被砍, 只挡伪造天价值。旧客户端不传 display_coin 时退回
+    「看满 10 秒发整份」(兼容不断币)。因子2(LT)由**客户端**按 granted 行 COUNT(拉自 /feed-reward/units)
+    算进 display_coin, 后端只记 granted 行让该计数自增, 不再服务端重算份值。
     - aborted=True(用户中途 ✕ 关闭这条):本条不发,记 status='closed_early'。
-    - 时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
+    - display_coin 为 0 且时长不足一份:记 status='too_short' 不发(不计 LT / 当日上限)。
     - 命中当日条数上限:记 status='capped' 不发。
-    duration_seconds 是**这一条**的观看秒数。服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS、
+    duration_seconds 落库留痕(unit_count 字段), 旧端兼容路径据它判是否满 1 份。
-    eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。
     feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算。
     ad_type:广告形态(feed 信息流 / draw Draw 信息流),仅归类落库;**每日上限与因子2(LT)仍按本表
     全表 unit 累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**。
@@ -139,14 +142,26 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 整场总时长不足 10 秒,凑不满一份 → 不发,记 too_short 留痕。
+    # 所见即所得(用户 2026-06-27「显示多少给多少」): 优先发**客户端小球显示**的金币 display_coin,
-    if unit_count == 0:
+    # 钳到本条「1 份满额」防刷(eCPM 已钳 AD_ECPM_MAX_FEN; 合法显示=因子2×p≤1份, 不会被砍)。
+    # 因子2(LT)按账号累计已发条数(granted 行 COUNT), 第 existing_ads+1 条。
+    existing_ads = granted_unit_total(db, user_id)
+    unit_cap = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
+    if display_coin > 0:
+        coin = min(display_coin, unit_cap)              # 新端: 所见即所得(直接发小球显示金币)
+    elif unit_count >= 1:
+        coin = unit_cap                                 # 旧端没传 display_coin: 退回「看满 1 份发整份」(兼容)
+    else:
+        coin = 0
+
+    # 显示金币为 0 且没满一份 → 不发, 记 too_short 留痕(不写 granted 行 → 不计 LT / 当日上限)。
+    if coin <= 0:
         rec = AdFeedRewardRecord(
             client_event_id=client_event_id,
             user_id=user_id,
             reward_date=today,
             duration_seconds=safe_duration,
-            unit_count=0,
+            unit_count=unit_count,
             ad_session_id=ad_session_id,
             ecpm_raw=ecpm,
             adn=adn,
@@ -161,15 +176,11 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    # 一条广告 = 一个「单次公式值」(因子2 按账号累计**条**数, 即第 existing_ads+1 条);看满一份(unit_count>=1)即发,不逐份累加。
+    crud_wallet.grant_coins(
-    existing_ads = granted_unit_total(db, user_id)
+        db, user_id, coin,
-    coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
+        biz_type="feed_ad_reward", ref_id=client_event_id,
-    if coin > 0:
+        remark="信息流广告奖励",
-        crud_wallet.grant_coins(
+    )
-            db, user_id, coin,
-            biz_type="feed_ad_reward", ref_id=client_event_id,
-            remark="信息流广告奖励",
-        )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,
         user_id=user_id,

app/schemas/ad.py

@@ -167,6 +167,11 @@ class FeedRewardIn(BaseModel):
     aborted: bool = Field(
         False, description="用户中途 ✕ 关闭广告(未走完比价):整场不发,记 closed_early"
     )
+    display_coin: int = Field(
+        0, ge=0,
+        description="客户端金币小球**本条显示**的金币(所见即所得):后端直接发这个数,钳到本条最大 1 份"
+        "满额防刷。缺省 0 = 旧客户端不传,退回服务端「看满 10 秒发整份」",
+    )
 
 
 class FeedRewardOut(BaseModel):

docs/api/admin-ad-revenue-report.md

@@ -27,8 +27,11 @@
   | `date_to` | string | =`date_from` | 结束日 北京时间 `YYYY-MM-DD`,**闭区间**;单日时与 `date_from` 相同 |
   | `user_id` | int | 全部 | 只看某用户;不传=所有用户 |
   | `ad_type` | string | 全部 | `reward_video` / `feed` / `draw`;不传=全部类型 |
+  | `feed_scene` | string | 全部 | `comparison`(比价)/ `coupon`(领券)/ `welfare`(福利);**全局筛选**,同时作用于明细 / 合计 / `daily`·`hourly` 趋势;不传=全部场景 |
   | `granularity` | string | `day` | `day`=按天 / `hour`=按小时(聚合键再加北京时间小时 0–23);**区间>1 天建议用 day** |
-  | `limit` | int(1~1000) | 500 | **展示**明细组数(截断;`total`/`total_*`/`daily` 按全量统计不受影响) |
+  | `limit` | int(1~1000) | 500 | **每页条数**(分页大小);`total`/`total_*`/`daily`/`hourly` 按全量统计不受分页影响 |
+  | `offset` | int(≥0) | 0 | 分页偏移(已跳过条数)=(页码−1)×`limit` |
+  | `sort` | string | `time` | 明细排序:`time`=按时间倒序(新→旧) / `ecpm`=按 eCPM 数值倒序 |
 
   约束:`date_to` 不早于 `date_from`、区间最长 **92 天**、日期须 `YYYY-MM-DD`,否则 `422`。
 
@@ -36,15 +39,18 @@
   | 字段 | 类型 | 说明 |
   |---|---|---|
   | `date_from` / `date_to` | string | 报表起止日期(闭区间) |
-  | `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受 `limit` 影响) |
+  | `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受分页影响) |
-  | `total` | int | 聚合组**总数**(全量,不受 `limit` 影响) |
+  | `hourly` | `AdRevenueHourly[]` | 按小时汇总序列(全量,供按小时趋势图;**仅 `granularity=hour` 时非空**;不受分页影响) |
-  | `truncated` | bool | 明细是否被 `limit` 截断 |
+  | `type_stats` | `{[ad_type]: AdRevenueTypeStat}` | 按广告类型(`ad_type`)小计(全量);前端取 `draw` / `reward_video` 做分类大盘 |
+  | `dau` | int \| null | 今日活跃用户数(复用大盘口径 `last_login_at`,今日登录过);**仅查询=今日单天时有值**,历史/多天为 `null` |
+  | `total` | int | 当前筛选下的**分页总条数**(全量,不受分页影响;= 前端分页器 total) |
+  | `truncated` | bool | 当前页之后是否还有更多事件(`len(events) > offset + limit`) |
   | `total_impressions` | int | 全量展示条数合计 |
   | `total_revenue_yuan` | float | 全量收益合计(元) |
   | `total_expected_coin` | int | 全量应发金币合计 |
   | `total_actual_coin` | int | 全量实发金币合计 |
   | `mismatch_count` | int | 应发≠实发的组数(=0 说明全部按公式发放) |
-  | `items` | `AdRevenueRow[]` | 聚合明细(按 日期→用户→类型→代码位 排序) |
+  | `items` | `AdRevenueRow[]` | 逐条广告事件(**按时间倒序:新→旧**);`limit`/`offset` 对全量做分页切片,返回当前页 |
 
 ### AdRevenueDaily(`daily[]` — 按天趋势)
 | 字段 | 类型 | 说明 |
@@ -55,6 +61,21 @@
 | `expected_coin` | int | 当天应发金币合计 |
 | `actual_coin` | int | 当天实发金币合计 |
 
+### AdRevenueHourly(`hourly[]` — 按小时趋势,仅 `granularity=hour` 时非空)
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `hour` | int | 北京时间小时 0–23 |
+| `impressions` | int | 该小时展示条数合计 |
+| `revenue_yuan` | float | 该小时预估收益合计(元) |
+| `expected_coin` | int | 该小时应发金币合计 |
+| `actual_coin` | int | 该小时实发金币合计 |
+
+### AdRevenueTypeStat(`type_stats[ad_type]` — 分广告类型小计,供大盘第二行)
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `impressions` | int | 该类型展示条数合计 |
+| `revenue_yuan` | float | 该类型预估收益合计(元);eCPM 由前端用 收益÷展示×1000 算 |
+
 ### AdRevenueRow(`items[]`)
 | 字段 | 类型 | 说明 |
 |---|---|---|