feat(launch_confirm): 新增启动确认样本列表内部接口(供 pricebot distill 回读)

补齐「启动确认窗自进化闭环」里 app-server 这一侧的读路径:pricebot 的 launch_confirm
LLM 兜底成功样本会 server→server 落到 app-server 的 launch_confirm_sample 表;本提交
新增「样本列表查询」内部接口, 供 pricebot 的 scripts/distill_launch_confirm.py 回读这些
样本、按「宿主包 × 文案变体」聚合沉淀出候选规则。

- app/api/internal/launch_confirm.py: 新增样本列表内部接口(server→server, 不走用户 JWT)
- app/repositories/launch_confirm_sample.py: 样本列表查询
- app/schemas/launch_confirm_sample.py: 列表响应 schema

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

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 份数**账号累计**
-  (与 ad_feed_reward._unit_reward_total 的 existing_units 一致;同样不按天重置,
-  复算需叠加本日之前的累计份数)。
+- 信息流:**每条 granted = 1 份**(与 ad_feed_reward.grant_feed_reward 同口径:看满一份即发该条
+  满额,**不按 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 拆分**——
-    故无论 scene 怎么筛展示,这里都遍历当日**全部**信息流记录维持 granted_units 累加;scene 只决定
+    **关键:LT 因子账号累计按全表 granted 条数累计(feed+draw 共享同一发奖池/上限),不按 ad_type 拆分**——
+    故无论 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_units: dict[int, int] = _feed_prior_granted_units(db, date=date, user_id=user_id)
+    # 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
+    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)
-            units = rec.unit_count
-            granted_units[rec.user_id] = existing + units
+            # 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
+            # nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
+            # (即便 scene 不匹配不展示也要 +1,保证序号与正式发奖一致)。
+            nth = granted_count.get(rec.user_id, 0) + 1
+            granted_count[rec.user_id] = nth
             if not keep:
                 continue
-            expected = sum(
-                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
+            expected = rewards.calculate_ad_reward_coin(rec.ecpm_raw, nth)
             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,
-                "lt_index_start": start,
-                "lt_index_end": end,
-                "lt_factor_start": rewards.ad_lt_factor(start) if start else None,
-                "lt_factor_end": rewards.ad_lt_factor(end) if end else None,
+                "units": 1,
+                "lt_index_start": nth,
+                "lt_index_end": nth,
+                "lt_factor_start": rewards.ad_lt_factor(nth),
+                "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 时每行带北京小时(由各自时间算)。
-    limit 只截断 items(事件明细),total 与 total_* / daily 在全量上统计,数字始终可信。
+    ad_type: None=全部 / reward_video / feed / draw。feed_scene: None=全部 /
+    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

@@ -5,7 +5,8 @@ user.last_login_at / comparison_record.status / withdraw_order.status)要加索
 """
 from __future__ import annotations
 
-from datetime import datetime, timedelta, timezone
+from datetime import date, datetime, time, timedelta, timezone
+from decimal import Decimal, InvalidOperation
 
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -13,12 +14,27 @@ from sqlalchemy.orm import Session
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.comparison import ComparisonRecord
+from app.models.coupon_state import CouponPromptEngagement
+from app.models.cps_order import CpsOrder
 from app.models.feedback import Feedback
+from app.models.savings import SavingsRecord
 from app.models.signin import SigninBoostRecord, SigninRecord
 from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
 _BEIJING = timezone(timedelta(hours=8))
+COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
+COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
+EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
+UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
+REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
+    *COUPON_REWARD_BIZ_TYPES,
+    *COMPARISON_REWARD_BIZ_TYPES,
+    *EXCLUDED_REWARD_BIZ_TYPES,
+    *UNCLASSIFIED_FEED_BIZ_TYPES,
+)
+MEITUAN_CPS_INVALID_STATUSES = ("4", "5")
+MEITUAN_CPS_SETTLED_STATUS = "6"
 
 
 def _beijing_today_start_utc() -> datetime:
@@ -28,20 +44,90 @@ def _beijing_today_start_utc() -> datetime:
     return start_bj.astimezone(timezone.utc)
 
 
-def dashboard_overview(db: Session) -> dict:
+def today_dau(db: Session) -> int:
+    """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。
+
+    广告收益报表复用这个函数;历史窗口 DAU 由 dashboard_overview 的 period 口径另算。
+    """
     today_start = _beijing_today_start_utc()
+    return int(
+        db.execute(select(func.count(User.id)).where(User.last_login_at >= today_start)).scalar_one()
+    )
+
+
+def _default_period_end() -> date:
+    """新版大盘不含今日,默认窗口结束日=北京时间昨天。"""
+    return datetime.now(_BEIJING).date() - timedelta(days=1)
+
+
+def _normalize_period(date_from: date | None, date_to: date | None) -> tuple[date, date]:
+    end = date_to or _default_period_end()
+    start = date_from or end
+    if start > end:
+        start, end = end, start
+    return start, end
+
+
+def _period_bounds(date_from: date, date_to: date) -> tuple[datetime, datetime, datetime, datetime]:
+    """返回同一北京自然日窗口的 UTC aware 边界和北京 naive 边界。
+
+    user.created_at / last_login_at 是 UTC aware 口径;比较/金币等历史上有北京 naive
+    写入,所以两套边界同时保留。
+    """
+    start_bj = datetime.combine(date_from, time.min, tzinfo=_BEIJING)
+    end_bj = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=_BEIJING)
+    start_utc = start_bj.astimezone(timezone.utc)
+    end_utc = end_bj.astimezone(timezone.utc)
+    return (
+        start_utc,
+        end_utc,
+        start_bj.replace(tzinfo=None),
+        end_bj.replace(tzinfo=None),
+    )
+
+
+def _date_range(date_from: date, date_to: date) -> list[date]:
+    days = (date_to - date_from).days
+    return [date_from + timedelta(days=i) for i in range(days + 1)]
+
+
+def _commission_rate_percent(raw: str | None) -> Decimal | None:
+    """美团 commissionRate 原值: "300"=3%, "10"=0.1%;也兼容 "3%"。"""
+    if raw is None:
+        return None
+    s = str(raw).strip()
+    if not s:
+        return None
+    try:
+        if s.endswith("%"):
+            return Decimal(s[:-1])
+        val = Decimal(s)
+    except (InvalidOperation, ValueError):
+        return None
+    return val / Decimal("100")
+
+
+def dashboard_overview(
+    db: Session, *, date_from: date | None = None, date_to: date | None = None
+) -> dict:
+    today_start = _beijing_today_start_utc()
+    period_from, period_to = _normalize_period(date_from, date_to)
+    start_utc, end_utc, start_local, end_local = _period_bounds(period_from, period_to)
 
     def _count(model, *conds) -> int:
         stmt = select(func.count(model.id))
         if conds:
             stmt = stmt.where(*conds)
-        return db.execute(stmt).scalar_one()
+        return int(db.execute(stmt).scalar_one())
 
     def _sum(col, *conds) -> int:
         stmt = select(func.coalesce(func.sum(col), 0))
         if conds:
             stmt = stmt.where(*conds)
-        return db.execute(stmt).scalar_one()
+        return int(db.execute(stmt).scalar_one())
+
+    def _user_id_set(stmt) -> set[int]:
+        return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}
 
     # ===== 用户 =====
     by_status = dict(
@@ -61,6 +147,204 @@ def dashboard_overview(db: Session) -> dict:
     comparison_total = _count(ComparisonRecord)
     comparison_success = _count(ComparisonRecord, ComparisonRecord.status == "success")
     success_rate = round(comparison_success / comparison_total, 4) if comparison_total else 0.0
+    period_comparison_conds = (
+        ComparisonRecord.created_at >= start_local,
+        ComparisonRecord.created_at < end_local,
+    )
+    period_comparison_total = _count(ComparisonRecord, *period_comparison_conds)
+    period_comparison_success = _count(
+        ComparisonRecord,
+        *period_comparison_conds,
+        ComparisonRecord.status == "success",
+    )
+    period_comparison_success_rate = (
+        round(period_comparison_success / period_comparison_total, 4)
+        if period_comparison_total
+        else 0.0
+    )
+    period_saved_positive_count = _count(
+        ComparisonRecord,
+        *period_comparison_conds,
+        ComparisonRecord.status == "success",
+        ComparisonRecord.saved_amount_cents > 0,
+    )
+    period_saved_positive_sum = _sum(
+        ComparisonRecord.saved_amount_cents,
+        *period_comparison_conds,
+        ComparisonRecord.status == "success",
+        ComparisonRecord.saved_amount_cents > 0,
+    )
+    period_avg_saved_cents = (
+        round(period_saved_positive_sum / period_saved_positive_count)
+        if period_saved_positive_count
+        else None
+    )
+    period_avg_duration_ms = db.execute(
+        select(func.avg(ComparisonRecord.total_ms)).where(
+            *period_comparison_conds,
+            ComparisonRecord.total_ms.is_not(None),
+            ComparisonRecord.total_ms > 0,
+        )
+    ).scalar_one()
+    period_avg_duration_ms = (
+        round(float(period_avg_duration_ms))
+        if period_avg_duration_ms is not None
+        else None
+    )
+
+    ordered_exists = (
+        select(SavingsRecord.id)
+        .where(
+            SavingsRecord.user_id == ComparisonRecord.user_id,
+            SavingsRecord.source == "compare",
+            SavingsRecord.shop_name.is_not(None),
+            SavingsRecord.shop_name == ComparisonRecord.store_name,
+        )
+        .exists()
+    )
+    period_ordered_count = _count(
+        ComparisonRecord,
+        *period_comparison_conds,
+        ComparisonRecord.store_name.is_not(None),
+        ordered_exists,
+    )
+
+    # ===== 日期窗口用户 =====
+    period_new_user_ids = _user_id_set(
+        select(User.id).where(User.created_at >= start_utc, User.created_at < end_utc)
+    )
+    login_user_ids = _user_id_set(
+        select(User.id).where(User.last_login_at >= start_utc, User.last_login_at < end_utc)
+    )
+    compare_user_ids = _user_id_set(
+        select(ComparisonRecord.user_id).where(*period_comparison_conds)
+    )
+    coupon_user_ids = _user_id_set(
+        select(CouponPromptEngagement.user_id).where(
+            CouponPromptEngagement.engage_date >= period_from,
+            CouponPromptEngagement.engage_date <= period_to,
+            CouponPromptEngagement.engage_type == "claim_started",
+        )
+    )
+    period_active_user_ids = login_user_ids | compare_user_ids | coupon_user_ids
+    period_retained_new_user_ids = period_new_user_ids & period_active_user_ids
+    period_retention_rate = (
+        round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4)
+        if period_new_user_ids
+        else None
+    )
+    trend_points: list[dict] = []
+    for cur_date in _date_range(period_from, period_to):
+        day_start_utc, day_end_utc, day_start_local, day_end_local = _period_bounds(
+            cur_date, cur_date
+        )
+        daily_comparison_conds = (
+            ComparisonRecord.created_at >= day_start_local,
+            ComparisonRecord.created_at < day_end_local,
+        )
+        daily_login_user_ids = _user_id_set(
+            select(User.id).where(
+                User.last_login_at >= day_start_utc,
+                User.last_login_at < day_end_utc,
+            )
+        )
+        daily_compare_user_ids = _user_id_set(
+            select(ComparisonRecord.user_id).where(*daily_comparison_conds)
+        )
+        daily_coupon_user_ids = _user_id_set(
+            select(CouponPromptEngagement.user_id).where(
+                CouponPromptEngagement.engage_date == cur_date,
+                CouponPromptEngagement.engage_type == "claim_started",
+            )
+        )
+        trend_points.append(
+            {
+                "date": cur_date,
+                "active_users": len(
+                    daily_login_user_ids | daily_compare_user_ids | daily_coupon_user_ids
+                ),
+                "new_users": _count(
+                    User,
+                    User.created_at >= day_start_utc,
+                    User.created_at < day_end_utc,
+                ),
+                "comparisons": _count(ComparisonRecord, *daily_comparison_conds),
+            }
+        )
+
+    period_coin_conds = (
+        CoinTransaction.created_at >= start_local,
+        CoinTransaction.created_at < end_local,
+        CoinTransaction.amount > 0,
+    )
+    period_reward_video_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
+    )
+    period_feed_ad_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type == "feed_ad_reward",
+    )
+    period_signin_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type == "signin",
+    )
+    period_signin_boost_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type == "signin_boost",
+    )
+    period_task_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.like("task_%"),
+    )
+    period_coupon_reward_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
+    )
+    period_comparison_reward_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
+    )
+    period_regular_task_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES),
+    )
+    period_meituan_orders = list(
+        db.execute(
+            select(CpsOrder).where(
+                CpsOrder.pay_time >= start_utc,
+                CpsOrder.pay_time < end_utc,
+            )
+        ).scalars()
+    )
+    period_meituan_valid_orders = [
+        o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES
+    ]
+    period_meituan_hit_count = 0
+    period_meituan_miss_count = 0
+    period_meituan_unknown_rate_count = 0
+    for order in period_meituan_valid_orders:
+        rate = _commission_rate_percent(order.commission_rate)
+        if rate is None:
+            period_meituan_unknown_rate_count += 1
+        elif rate < Decimal("1"):
+            period_meituan_miss_count += 1
+        else:
+            period_meituan_hit_count += 1
+    period_meituan_hit_denominator = period_meituan_hit_count + period_meituan_miss_count
+    period_meituan_hit_rate = (
+        round(period_meituan_hit_count / period_meituan_hit_denominator, 4)
+        if period_meituan_hit_denominator
+        else None
+    )
 
     return {
         "users": {
@@ -69,7 +353,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 之和;负数是兑换/扣减不计)
@@ -119,7 +403,61 @@ def dashboard_overview(db: Session) -> dict:
             "success": comparison_success,
             "success_rate": success_rate,
         },
-        "feedback": {"new": _count(Feedback, Feedback.status.in_(("pending", "new")))},
-        # CPS 收入数据源未接(referral-link 只换链接,转化/佣金未回收)→ 前端显示"待接入"。
-        "cps": {"available": False, "note": "CPS 转化数据未接入(P2)"},
+        "period": {
+            "date_from": period_from,
+            "date_to": period_to,
+            "users": {
+                "new": len(period_new_user_ids),
+                "active": len(period_active_user_ids),
+                "retained_new_users": len(period_retained_new_user_ids),
+                "retention_rate": period_retention_rate,
+                "retention_note": (
+                    "近似口径:登录(last_login_at)+已上报比价记录+领券claim_started;"
+                    "尚不包含未完成上报的比价开始事件"
+                ),
+            },
+            "comparison": {
+                "total": period_comparison_total,
+                "success": period_comparison_success,
+                "success_rate": period_comparison_success_rate,
+                "ordered": period_ordered_count,
+                "average_duration_ms": period_avg_duration_ms,
+                "average_saved_cents": period_avg_saved_cents,
+            },
+            "coins": {
+                "granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
+                "reward_video_coin_total": period_reward_video_coin_total,
+                "feed_ad_coin_total": period_feed_ad_coin_total,
+                "signin_coin_total": period_signin_coin_total,
+                "signin_boost_coin_total": period_signin_boost_coin_total,
+                "task_coin_total": period_task_coin_total,
+                "coupon_reward_coin_total": period_coupon_reward_coin_total,
+                "comparison_reward_coin_total": period_comparison_reward_coin_total,
+                "regular_task_coin_total": period_regular_task_coin_total,
+            },
+            "cash": {
+                "withdraw_success_cents": _sum(
+                    WithdrawOrder.amount_cents,
+                    WithdrawOrder.status == "success",
+                    WithdrawOrder.created_at >= start_local,
+                    WithdrawOrder.created_at < end_local,
+                ),
+            },
+            "trend": trend_points,
+        },
+        "feedback": {
+            "new": _count(Feedback, Feedback.status.in_(("pending", "new"))),
+        },
+        "cps": {
+            "available": True,
+            "note": "美团 CPS 读 cps_order 对账订单;淘宝/京东佣金暂空",
+            "meituan_order_count": len(period_meituan_valid_orders),
+            "meituan_commission_cents": sum(
+                o.commission_cents or 0 for o in period_meituan_valid_orders
+            ),
+            "meituan_hit_count": period_meituan_hit_count,
+            "meituan_miss_count": period_meituan_miss_count,
+            "meituan_unknown_rate_count": period_meituan_unknown_rate_count,
+            "meituan_hit_rate": period_meituan_hit_rate,
+        },
     }

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/routers/cps.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import time
-from datetime import datetime, timedelta, timezone
+from datetime import date as _date, datetime, time as _dt_time, timedelta, timezone
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
@@ -340,24 +340,77 @@ def generate_referral_links(
 
 
 # ───────────── 订单对账 ─────────────
+_BEIJING = timezone(timedelta(hours=8))
+
+
+def _parse_day(value: str | None, *, field: str) -> _date | None:
+    if value is None:
+        return None
+    try:
+        return _date.fromisoformat(value)
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e
+
+
+def _reconcile_range_to_ts(
+    date_from: _date | None, date_to: _date | None, days: int
+) -> tuple[int, int]:
+    if date_from is None and date_to is None:
+        now = int(time.time())
+        return now - days * 86400, now
+
+    start_day = date_from or date_to
+    end_day = date_to or date_from
+    if start_day is None or end_day is None:
+        raise HTTPException(status_code=422, detail="日期参数不完整")
+    if start_day > end_day:
+        start_day, end_day = end_day, start_day
+    if (end_day - start_day).days + 1 > 90:
+        raise HTTPException(status_code=422, detail="美团订单查询最长 90 天")
+
+    start_dt = datetime.combine(start_day, _dt_time.min, tzinfo=_BEIJING)
+    end_dt = datetime.combine(end_day + timedelta(days=1), _dt_time.min, tzinfo=_BEIJING)
+    return int(start_dt.timestamp()), int(end_dt.timestamp())
+
+
 @router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账")
 def reconcile_orders(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("finance"))],
     db: AdminDb,
-    days: Annotated[int, Query(ge=1, le=90)] = 7,
+    date_from: Annotated[str | None, Query(description="起始日 YYYY-MM-DD")] = None,
+    date_to: Annotated[str | None, Query(description="结束日 YYYY-MM-DD")] = None,
+    days: Annotated[int, Query(ge=1, le=90, description="未传日期时默认拉近 N 天")] = 7,
     sid: Annotated[str | None, Query(max_length=64)] = None,
+    query_time_type: Annotated[int, Query(ge=1, le=2)] = 1,
 ) -> CpsReconcileResult:
-    now = int(time.time())
+    start_ts, end_ts = _reconcile_range_to_ts(
+        _parse_day(date_from, field="date_from"),
+        _parse_day(date_to, field="date_to"),
+        days,
+    )
     try:
         result = cps_repo.reconcile_orders(
-            db, start_time=now - days * 86400, end_time=now, sid=sid,
+            db,
+            start_time=start_ts,
+            end_time=end_ts,
+            query_time_type=query_time_type,
+            sid=sid,
         )
     except MeituanCpsError as e:
         raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e
     write_audit(
         db, admin, action="cps.order.reconcile", target_type="cps_order", target_id=None,
-        detail={"days": days, "sid": sid, **result}, ip=get_client_ip(request), commit=True,
+        detail={
+            "date_from": date_from,
+            "date_to": date_to,
+            "days": days,
+            "sid": sid,
+            "query_time_type": query_time_type,
+            **result,
+        },
+        ip=get_client_ip(request),
+        commit=True,
     )
     return CpsReconcileResult(**result)
 

app/admin/routers/dashboard.py

@@ -1,7 +1,9 @@
 """admin 数据大盘(只读聚合)。"""
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends
+from datetime import date
+
+from fastapi import APIRouter, Depends, Query
 
 from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import stats
@@ -15,5 +17,11 @@ router = APIRouter(
 
 
 @router.get("/overview", response_model=DashboardOverview, summary="大盘核心指标")
-def overview(db: AdminDb) -> DashboardOverview:
-    return DashboardOverview.model_validate(stats.dashboard_overview(db))
+def overview(
+    db: AdminDb,
+    date_from: date | None = Query(None, description="北京时间自然日起始日 YYYY-MM-DD"),
+    date_to: date | None = Query(None, description="北京时间自然日结束日 YYYY-MM-DD"),
+) -> DashboardOverview:
+    return DashboardOverview.model_validate(
+        stats.dashboard_overview(db, date_from=date_from, date_to=date_to)
+    )

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 影响)")
-    truncated: bool = Field(..., description="明细是否被 limit 截断")
+    hourly: list[AdRevenueHourly] = Field(
+        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/admin/schemas/dashboard.py

@@ -1,6 +1,8 @@
 """admin 大盘 schemas(对应 stats.dashboard_overview 的嵌套结构)。"""
 from __future__ import annotations
 
+from datetime import date
+
 from pydantic import BaseModel
 
 
@@ -38,6 +40,56 @@ class DashboardComparison(BaseModel):
     success_rate: float
 
 
+class DashboardPeriodUsers(BaseModel):
+    new: int
+    active: int
+    retained_new_users: int
+    retention_rate: float | None = None
+    retention_note: str
+
+
+class DashboardPeriodComparison(BaseModel):
+    total: int
+    success: int
+    success_rate: float
+    ordered: int
+    average_duration_ms: int | None = None
+    average_saved_cents: int | None = None
+
+
+class DashboardPeriodCoins(BaseModel):
+    granted_total: int
+    reward_video_coin_total: int = 0
+    feed_ad_coin_total: int = 0
+    signin_coin_total: int = 0
+    signin_boost_coin_total: int = 0
+    task_coin_total: int = 0
+    coupon_reward_coin_total: int = 0
+    comparison_reward_coin_total: int = 0
+    regular_task_coin_total: int = 0
+
+
+class DashboardPeriodCash(BaseModel):
+    withdraw_success_cents: int
+
+
+class DashboardTrendPoint(BaseModel):
+    date: date
+    active_users: int
+    new_users: int
+    comparisons: int
+
+
+class DashboardPeriod(BaseModel):
+    date_from: date
+    date_to: date
+    users: DashboardPeriodUsers
+    comparison: DashboardPeriodComparison
+    coins: DashboardPeriodCoins
+    cash: DashboardPeriodCash
+    trend: list[DashboardTrendPoint] = []
+
+
 class DashboardFeedback(BaseModel):
     new: int
 
@@ -45,6 +97,12 @@ class DashboardFeedback(BaseModel):
 class DashboardCps(BaseModel):
     available: bool
     note: str
+    meituan_order_count: int = 0
+    meituan_commission_cents: int = 0
+    meituan_hit_count: int = 0
+    meituan_miss_count: int = 0
+    meituan_unknown_rate_count: int = 0
+    meituan_hit_rate: float | None = None
 
 
 class DashboardOverview(BaseModel):
@@ -52,5 +110,6 @@ class DashboardOverview(BaseModel):
     coins: DashboardCoins
     cash: DashboardCash
     comparison: DashboardComparison
+    period: DashboardPeriod
     feedback: DashboardFeedback
     cps: DashboardCps

app/api/internal/launch_confirm.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import hmac
 import logging
+from datetime import datetime, timedelta, timezone
 from typing import Annotated
 
 from fastapi import APIRouter, Header, HTTPException, status
@@ -21,6 +22,7 @@ from app.repositories import launch_confirm_sample as repo
 from app.schemas.launch_confirm_sample import (
     LaunchConfirmSampleIn,
     LaunchConfirmSampleOut,
+    LaunchConfirmSampleRow,
 )
 
 logger = logging.getLogger("shagua.internal.launch_confirm")
@@ -61,3 +63,34 @@ def report_launch_confirm_sample(
         payload.exec_success, payload.trace_id,
     )
     return LaunchConfirmSampleOut(id=sid)
+
+
+@router.get(
+    "/launch-confirm-samples",
+    response_model=list[LaunchConfirmSampleRow],
+    summary="启动确认窗兜底样本列表(沉淀脚本 distill 读; server→server, 不走 JWT)",
+)
+def list_launch_confirm_samples(
+    db: DbSession,
+    x_internal_secret: Annotated[str | None, Header()] = None,
+    exec_success: bool | None = None,
+    host_package: str | None = None,
+    since_days: int | None = None,
+    limit: int = 1000,
+) -> list[LaunchConfirmSampleRow]:
+    """读样本供 pricebot 的 distill_launch_confirm.py 聚合沉淀回 PROFILES。
+
+    与上报端点同一把共享密钥;exec_success/host_package/since_days 均可选,limit 默认 1000。
+    """
+    _check_secret(x_internal_secret)
+    since = None
+    if since_days and since_days > 0:
+        since = datetime.now(timezone.utc) - timedelta(days=since_days)
+    rows = repo.list_samples(
+        db,
+        exec_success=exec_success,
+        host_package=host_package,
+        since=since,
+        limit=limit,
+    )
+    return [LaunchConfirmSampleRow.model_validate(r) for r in rows]

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:
-            resp = await client.post(
-                url, content=raw, headers={"Content-Type": "application/json"}
-            )
+        client = get_pricebot_client()
+        resp = await client.post(
+            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:
-            resp = await client.post(
-                url, content=raw, headers={"Content-Type": "application/json"}
-            )
+        client = get_pricebot_client()
+        resp = await client.post(
+            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/models/__init__.py

@@ -19,6 +19,7 @@ from app.models.coupon_state import (  # noqa: F401
     CouponDailyCompletion,
     CouponPromptEngagement,
 )
+from app.models.cps_order import CpsOrder  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401
 from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401

app/repositories/ad_ecpm.py

@@ -53,11 +53,16 @@ def create_ecpm_record(
         db.commit()
     except IntegrityError:
         db.rollback()
-        if ad_session_id:
-            existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
-            if existing is not None:
-                return existing
-        raise
+        # 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
+        # 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
+        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
+        # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
+        # 旧逻辑在此 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)按账号累计
-    **条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。
+    发奖规则(所见即所得, 2026-06-27 用户拍板「显示多少给多少」):优先**直接发客户端小球显示的金币
+    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、
-    eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。
+    duration_seconds 落库留痕(unit_count 字段), 旧端兼容路径据它判是否满 1 份。
     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 留痕。
-    if unit_count == 0:
+    # 所见即所得(用户 2026-06-27「显示多少给多少」): 优先发**客户端小球显示**的金币 display_coin,
+    # 钳到本条「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)即发,不逐份累加。
-    existing_ads = granted_unit_total(db, user_id)
-    coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1)
-    if coin > 0:
-        crud_wallet.grant_coins(
-            db, user_id, coin,
-            biz_type="feed_ad_reward", ref_id=client_event_id,
-            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/repositories/launch_confirm_sample.py

@@ -2,8 +2,10 @@
 from __future__ import annotations
 
 import logging
+from datetime import datetime
 from typing import Optional
 
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.models.launch_confirm_sample import LaunchConfirmSample
@@ -33,3 +35,27 @@ def insert_sample(db: Session, payload: LaunchConfirmSampleIn) -> int:
     db.commit()
     db.refresh(row)
     return row.id
+
+
+def list_samples(
+    db: Session,
+    *,
+    exec_success: Optional[bool] = None,
+    host_package: Optional[str] = None,
+    since: Optional[datetime] = None,
+    limit: int = 1000,
+) -> list[LaunchConfirmSample]:
+    """按条件查样本(pricebot 的 distill_launch_confirm.py 沉淀工具读)。created_at 升序。
+
+    过滤项都可空:exec_success(只看放行成功的)/ host_package(只看某宿主包)/
+    since(>= created_at)。limit 兜底防一次拉爆全表。
+    """
+    stmt = select(LaunchConfirmSample)
+    if exec_success is not None:
+        stmt = stmt.where(LaunchConfirmSample.exec_success == exec_success)
+    if host_package:
+        stmt = stmt.where(LaunchConfirmSample.host_package == host_package)
+    if since is not None:
+        stmt = stmt.where(LaunchConfirmSample.created_at >= since)
+    stmt = stmt.order_by(LaunchConfirmSample.created_at.asc()).limit(limit)
+    return list(db.execute(stmt).scalars().all())

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):

app/schemas/compare_record.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 # ===== 上报请求 =====
@@ -21,8 +21,36 @@ class ComparisonItemIn(BaseModel):
 
     name: str
     qty: int = 1
+    # specs 仅供比价记录展示(admin-web 详情 / app 记录页都按字符串数组渲染并 join)。
+    # pricebot(2026-06-27 嵌套规格统一)起 calibration 的 specs 可能是规格对象
+    # [{name, qty, sub_specs}] 而非字符串 → 下面的 before-validator 统一拍平成可读字符串,
+    # 兼容新旧两种输入、保持 list[str] 契约不变(下游零改动)。
+    # ⚠️ 两个坑都踩过, 必须"拍平"而非别的: ① 直接声明 list[str] 不拍平 → 对象 specs 整条
+    # 422 被拒、不入库(同下方 platform_results list→dict 同类事故); ② 仅放宽成裸 list 又会让
+    # 下游 join 出 "[object Object]"/对象 toString 的乱码。
     specs: list[str] | None = None
 
+    @field_validator("specs", mode="before")
+    @classmethod
+    def _flatten_specs(cls, v: object) -> object:
+        """pricebot 规格对象 [{name, qty, sub_specs}] → 可读字符串数组; 字符串元素原样保留;
+        非 list 原样交还(让 pydantic 照常报类型错)。嵌套规格拼成 '主项(子1,子2)'。"""
+        if not isinstance(v, list):
+            return v
+        out: list[str] = []
+        for it in v:
+            if isinstance(it, str):
+                s = it
+            elif isinstance(it, dict):
+                name = str(it.get("name") or "").strip()
+                subs = [str(x).strip() for x in (it.get("sub_specs") or []) if str(x).strip()]
+                s = f"{name}({','.join(subs)})" if name and subs else (name or ",".join(subs))
+            else:
+                continue
+            if s:
+                out.append(s)
+        return out
+
 
 class AppliedCouponIn(BaseModel):
     """单笔已用优惠(来自 comparison_results[].applied_coupons)。amount 单位:元、正数。"""
@@ -95,6 +123,7 @@ class ComparisonRecordIn(BaseModel):
     # pricebot done.params.trace_url 原样上报,落库供记录页「复制调试链接」(dir 名含落盘
     # 时分秒前端拼不出,必须由后端透传)。
     trace_url: str | None = Field(None, description="本次比价公网调试链接")
+    total_ms: int | None = Field(None, description="整场比价墙钟耗时(ms)")
 
     # ===== debug 维度(客户端采集上报;旧客户端不带 → None。仅 admin 比价记录页用)=====
     # 必须显式声明,否则 model_dump() 落 raw_payload 时被 pydantic 静默丢弃(同上面 coupon_saved 的坑)。
@@ -144,6 +173,7 @@ class ComparisonRecordOut(BaseModel):
     items: list = []
     comparison_results: list = []
     skipped_dish_names: list = []
+    total_ms: int | None = None
     # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
     # 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。
     ordered: bool = False

app/schemas/launch_confirm_sample.py

@@ -1,7 +1,9 @@
 """启动确认窗兜底样本的内部上报模型(pricebot → app-server)。"""
 from __future__ import annotations
 
-from pydantic import BaseModel
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
 
 
 class LaunchConfirmSampleIn(BaseModel):
@@ -24,3 +26,20 @@ class LaunchConfirmSampleOut(BaseModel):
     """落库结果。"""
 
     id: int
+
+
+class LaunchConfirmSampleRow(BaseModel):
+    """单条样本(列表读出;沉淀脚本 distill 用,payload 原样带出)。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    created_at: datetime
+    trace_id: str | None = None
+    device_id: str | None = None
+    host_package: str | None = None
+    target_app: str | None = None
+    system_locale: str | None = None
+    exec_success: bool = False
+    dialog_title: str | None = None
+    payload: dict | None = None

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` 影响) |
-  | `total` | int | 聚合组**总数**(全量,不受 `limit` 影响) |
-  | `truncated` | bool | 明细是否被 `limit` 截断 |
+  | `daily` | `AdRevenueDaily[]` | 按日期汇总序列(全量,供按天趋势图;不受分页影响) |
+  | `hourly` | `AdRevenueHourly[]` | 按小时汇总序列(全量,供按小时趋势图;**仅 `granularity=hour` 时非空**;不受分页影响) |
+  | `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[]`)
 | 字段 | 类型 | 说明 |
 |---|---|---|

docs/guides/邀请功能-实现原理与本地测试.md

@@ -37,7 +37,7 @@ B 安装并首启 App
         └─ POST /api/v1/invite/bind { invite_code, channel="clipboard" }
         ↓
 后端 repositories/invite.py bind()
-  └─ 过四道防线 → 建 invite_relation + 给 A、B 各发金币(同事务原子提交)
+  └─ 过四道防线 → 建 invite_relation(邀请金币已下线,不写金币流水)
 ```
 
 手动填码这条:B 在邀请页输码 → `InviteRepository.bindManual()` → `POST /bind { channel="manual" }` → 同一个 `bind()`。
@@ -52,20 +52,20 @@ B 安装并首启 App
 |---|---|
 | 端点 | `app/api/v1/invite.py`:`GET /api/v1/invite/me`(返回 `invite_code` + `share_url` + 战绩)、`POST /api/v1/invite/bind`(绑定,`channel` = `clipboard` / `manual`)。**均需 Bearer 鉴权**。 |
 | share_url 构造 | `invite.py` 的 `my_invite`:`settings.INVITE_LANDING_URL + "?ref=" + code`。`INVITE_LANDING_URL` 在 `app/core/config.py`(默认 `https://app-api.shaguabijia.com/media/dl.html`)。 |
-| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 累计金币)。 |
+| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 兼容累计金币字段,当前恒为 0)。 |
 | 数据模型 | `app/models/invite.py` 的 `InviteRelation`(`inviter_user_id` / `invitee_user_id` / `channel` / `status` / `inviter_coin` / `invitee_coin` / `created_at`)+ `app/models/user.py` 的 `User.invite_code` 列。 |
 | 迁移 | `alembic/versions/invite_code_and_relation.py`:给 `user` 加 `invite_code`(唯一索引)+ 建 `invite_relation` 表。`down_revision = 11a1d08c6f55`。 |
 | 收发模型 | `app/schemas/invite.py`:`InviteInfoOut` / `BindInviteIn` / `BindInviteOut`。 |
-| 奖励常量 | `app/core/rewards.py`:`INVITE_INVITER_COINS` / `INVITE_INVITEE_COINS`(各 10000 = 1 元)、`INVITE_NEW_USER_WINDOW_HOURS`(72)。 |
+| 新人窗口 | `app/core/rewards.py`:`INVITE_NEW_USER_WINDOW_HOURS`(72)。邀请金币已下线,不再配置邀请金币常量。 |
 
 **`bind()` 的四道防线(防重复 / 防刷,看 `repositories/invite.py`):**
 
-1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复发奖)。
+1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复绑定)。
 2. **自邀屏蔽**:`inviter == invitee` → `self_invite`。
-3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才发奖,挡存量老用户互相填码薅羊毛 → 否则 `not_eligible`。
+3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才生效,挡存量老用户互相填码刷关系 → 否则 `not_eligible`。
 4. **手机号唯一**(天然限量):每个 B = 一个真实手机号账号。
 
-发金币复用 `repositories/wallet.py` 的 `grant_coins`,与建关系记录在**同一事务**提交,保证"建关系 + 双方加金币"原子。
+邀请金币已下线:`bind()` 只记录绑定关系,不再写 `coin_transaction`;响应里的金币字段保留兼容旧客户端,当前恒为 0。
 
 ### 3.2 前端(shaguabijia-app-android)
 
@@ -120,7 +120,7 @@ B 安装并首启 App
 ### 4.4 测试硬约束 / 坑(都是机制,不是 bug)
 
 - **B 必须用新手机号**:`invitee_user_id` 唯一,一个 B 只能绑一次;反复测要换号(或手删 `invite_relation` 那行 + 回滚金币)。
-- **72h 新人闸**:B 注册后 72 小时内绑才发奖(刚注册肯定满足)。
+- **72h 新人闸**:B 注册后 72 小时内绑定才生效(刚注册肯定满足)。
 - **A ≠ B**:自邀被屏蔽。
 - **B 从点下载到首启 App 之间别复制别的东西**:剪贴板会被覆盖 → 归因丢(剪贴板 deferred deeplink 的固有脆弱性)。
 - **笔记本 IP 别变**:debug 包把 `BASE_URL` 的 IP 烧死在编译期,DHCP 一换就连不上 → 给笔记本固定个 LAN IP。

h5/shared/api.js

@@ -0,0 +1,52 @@
+/**
+ * SGApi —— H5 调 app-server 后端的薄封装。
+ *
+ * 同源：H5 由 app-server 的 /media/h5/ 托管，后端在 /api/v1，同 host:port → 用相对路径、无 CORS。
+ * 鉴权：JWT Bearer。token 经 SGBridge.getToken() 从原生取(原生持登录态)；浏览器调试走 bridge 的 mock token。
+ * 401：交原生拉登录(requestLogin)兜底，本次请求按失败 reject；正式的 refresh 重试策略阶段2 再补。
+ *
+ * 依赖 shared/bridge.js 先加载(取 token)。
+ */
+(function (global) {
+  'use strict';
+
+  var BASE = '/api/v1';
+
+  function authHeaders() {
+    var t = (global.SGBridge && global.SGBridge.getToken()) || '';
+    var h = { 'Content-Type': 'application/json' };
+    if (t) h['Authorization'] = 'Bearer ' + t;
+    return h;
+  }
+
+  function handle(res) {
+    if (res.status === 401) {
+      // 未授权：拉原生登录(异步)，本次请求按失败处理，调用方自行决定是否重试
+      if (global.SGBridge) global.SGBridge.requestLogin();
+      return Promise.reject(new Error('unauthorized'));
+    }
+    if (!res.ok) {
+      return res.text().then(function (t) {
+        return Promise.reject(new Error('http ' + res.status + ' ' + t));
+      });
+    }
+    // 204 / 空体兜底
+    return res.text().then(function (t) { return t ? JSON.parse(t) : null; });
+  }
+
+  /** GET /api/v1<path>。path 以 / 开头，如 '/savings/battle'。 */
+  function apiGet(path) {
+    return fetch(BASE + path, { method: 'GET', headers: authHeaders() }).then(handle);
+  }
+
+  /** POST /api/v1<path>，body 自动 JSON 序列化。 */
+  function apiPost(path, body) {
+    return fetch(BASE + path, {
+      method: 'POST',
+      headers: authHeaders(),
+      body: JSON.stringify(body || {}),
+    }).then(handle);
+  }
+
+  global.SGApi = { base: BASE, get: apiGet, post: apiPost };
+})(window);

h5/shared/bridge.js

@@ -0,0 +1,162 @@
+/**
+ * SGBridge —— 傻瓜比价 H5 ↔ Android 原生 的桥。
+ *
+ * 背景：四个主 tab(首页/福利/记录/我的)由原生 Compose 改造为 WebView 加载本工程 H5。
+ * H5 只负责"画 + 取后端数据"；凡需要原生能力(登录态 / 跳转 / 跳外卖 App / 比价领券 /
+ * 权限 / 定位 / Toast / 激励视频)，一律经本桥调用原生。
+ *
+ * 协议两个方向：
+ *  ① H5 → 原生：Android 端 WebView.addJavascriptInterface(obj, "SGBridgeNative")。
+ *     obj 方法都是【同步】：查询类返回 String(JSON 或纯串)，动作类无返回。
+ *  ② 原生 → H5：原生执行 evaluateJavascript("window.SGBridge._emit('<event>', '<json>')")。
+ *     事件：onAuthChange(登录态变) / onBalanceChange(余额变) / onSigninChange(签到态变) / onResume(回前台刷新)。
+ *
+ * 离线兜底：浏览器里(无 SGBridgeNative)走 MOCK，便于不装 App 直接在本地 server 调样式 / 渲染。
+ * 与原生实现对应：见 shaguabijia-app-android 的 SGBridge.kt(方法名逐个对齐本文件)。
+ */
+(function (global) {
+  'use strict';
+
+  var native = global.SGBridgeNative || null;
+  var hasNative = !!native;
+
+  // ---- 离线 MOCK(仅无原生时生效，便于浏览器调试渲染；真机一律走 native) ----
+  var MOCK = {
+    authState: { loggedIn: true, userId: 1, nickname: '冰', avatarUrl: '', phone: '188****8888' },
+    token: 'mock-token-for-browser-debug',
+    deviceId: 'browser-debug-device',
+    appVersion: '0.0.0-debug',
+  };
+
+  function safeParse(s, fallback) {
+    try { return s ? JSON.parse(s) : fallback; } catch (e) { return fallback; }
+  }
+
+  // ====== 查询类(同步返回) ======
+
+  /** 当前登录态 + 用户基本信息 → {loggedIn, userId, nickname, avatarUrl, phone}。 */
+  function getAuthState() {
+    if (hasNative && native.getAuthState) return safeParse(native.getAuthState(), { loggedIn: false });
+    return MOCK.authState;
+  }
+
+  /** 后端鉴权用的 access token(空串=未登录)。原生持登录态，H5 调后端前取它拼 Bearer。 */
+  function getToken() {
+    if (hasNative && native.getToken) return native.getToken() || '';
+    return MOCK.token;
+  }
+
+  /** 设备唯一标识(心跳 / 领券状态查询等用)。 */
+  function getDeviceId() {
+    if (hasNative && native.getDeviceId) return native.getDeviceId() || '';
+    return MOCK.deviceId;
+  }
+
+  /** App 版本号。 */
+  function getAppVersion() {
+    if (hasNative && native.getAppVersion) return native.getAppVersion() || '';
+    return MOCK.appVersion;
+  }
+
+  /** 已安装的目标电商/外卖 App 包名数组(原生 InstalledApps 探测)。H5 选平台弹窗据此判真实装机态。 */
+  function getInstalledApps() {
+    if (hasNative && native.getInstalledApps) return safeParse(native.getInstalledApps(), []);
+    // MOCK(浏览器无原生):给主流已装,便于本地预览选平台弹窗正常显示"有"。
+    return ['com.sankuai.meituan', 'com.taobao.taobao', 'com.jingdong.app.mall', 'me.ele'];
+  }
+
+  /** 今日是否已领券(置灰「去领取」→「去查看」)。原生读 CompareButtonState(SP 按天);无桥默认 false。 */
+  function getCouponClaimedToday() {
+    if (hasNative && native.getCouponClaimedToday) return !!native.getCouponClaimedToday();
+    return false;
+  }
+
+  // ====== 动作类(无返回；异步结果走事件) ======
+
+  /** 跳原生页。route 取值对齐安卓 Routes(invite / settings / feedback / withdrawal / compareRecords / reportFlow / guideVideo / compareResult / coinHistory / cashHistory / welfareRules ...)。 */
+  function navigate(route) {
+    if (hasNative && native.navigate) native.navigate(route);
+    else console.log('[SGBridge mock] navigate →', route);
+  }
+
+  /** 拉起极光一键登录。结果异步经 onAuthChange 事件回来(不在此函数返回)。 */
+  function requestLogin() {
+    if (hasNative && native.requestLogin) native.requestLogin();
+    else console.log('[SGBridge mock] requestLogin');
+  }
+
+  /** 原生居中 Toast。 */
+  function toast(msg) {
+    if (hasNative && native.toast) native.toast(String(msg));
+    else console.log('[SGBridge mock] toast →', msg);
+  }
+
+  /** 跳美团/外卖 App(deeplink 优先；空则原生按包名启动，未装可跳应用商店)。 */
+  function openMeituan(deeplink) {
+    if (hasNative && native.openMeituan) native.openMeituan(deeplink || '');
+    else console.log('[SGBridge mock] openMeituan →', deeplink);
+  }
+
+  /** 触发 agent 比价流程(原生起无障碍引擎)。 */
+  function startCompare() {
+    if (hasNative && native.startCompare) native.startCompare();
+    else console.log('[SGBridge mock] startCompare');
+  }
+
+  /** 触发一键领券(原生先校验悬浮窗/无障碍权限，再起前台服务)。platforms: string[]。 */
+  function startCouponClaim(platforms) {
+    var json = JSON.stringify(platforms || []);
+    if (hasNative && native.startCouponClaim) native.startCouponClaim(json);
+    else console.log('[SGBridge mock] startCouponClaim →', json);
+  }
+
+  /** 按候选包名拉起对应平台 App。照原生 HomePicker 点击跳转:原生在该平台一组候选包里取首个可
+   *  getLaunchIntentForPackage 的拉起(NEW_TASK|CLEAR_TASK 冷启到平台首页)。packages: string[](一个平台一组候选包,任一可拉即拉)。 */
+  function launchApp(packages) {
+    var json = JSON.stringify(packages || []);
+    if (hasNative && native.launchApp) native.launchApp(json);
+    else console.log('[SGBridge mock] launchApp →', json);
+  }
+
+  // ====== 原生 → H5 事件总线 ======
+
+  var listeners = {}; // event → [fn]
+
+  /** 订阅原生事件。返回取消订阅函数。 */
+  function on(event, fn) {
+    (listeners[event] || (listeners[event] = [])).push(fn);
+    return function off() {
+      listeners[event] = (listeners[event] || []).filter(function (f) { return f !== fn; });
+    };
+  }
+
+  /** 供原生回调：window.SGBridge._emit('onAuthChange', '{...}')。payload 可为 JSON 串或对象。 */
+  function _emit(event, payload) {
+    var data = typeof payload === 'string' ? safeParse(payload, payload) : payload;
+    (listeners[event] || []).forEach(function (fn) {
+      try { fn(data); } catch (e) { console.error('[SGBridge] listener error', event, e); }
+    });
+  }
+
+  global.SGBridge = {
+    hasNative: hasNative,
+    // 查询
+    getAuthState: getAuthState,
+    getToken: getToken,
+    getDeviceId: getDeviceId,
+    getAppVersion: getAppVersion,
+    getInstalledApps: getInstalledApps,
+    getCouponClaimedToday: getCouponClaimedToday,
+    // 动作
+    navigate: navigate,
+    requestLogin: requestLogin,
+    toast: toast,
+    openMeituan: openMeituan,
+    startCompare: startCompare,
+    startCouponClaim: startCouponClaim,
+    launchApp: launchApp,
+    // 事件
+    on: on,
+    _emit: _emit,
+  };
+})(window);

tests/test_ad_reward.py

@@ -305,11 +305,12 @@ def test_feed_reward_grants_by_10_second_units(client) -> None:
         "duration_seconds": 30,
         "adn": "pangle",
         "slot_id": "slot_feed",
+        "display_coin": 4,
     }
     r = client.post("/api/v1/ad/feed-reward", json=payload, headers=_auth(token))
     assert r.status_code == 200, r.text
     body = r.json()
-    expected = sum(calculate_ad_reward_coin("200", i) for i in range(1, 4))
+    expected = 4
     assert body["granted"] is True
     assert body["status"] == "granted"
     assert body["unit_count"] == 3

tests/test_admin_read.py

@@ -64,7 +64,8 @@ def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
     assert data["users"]["total"] >= 1
     assert data["coins"]["granted_total"] >= 5000
     assert "success_rate" in data["comparison"]
-    assert data["cps"]["available"] is False
+    assert data["cps"]["available"] is True
+    assert "meituan_order_count" in data["cps"]
 
 
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:

tests/test_compare_record.py

@@ -44,6 +44,7 @@ def _food_payload(trace_id: str) -> dict:
         "skipped_dish_names": ["黑牛肉卷"],
         "total_dish_count": 3,
         "information": "在美团找到同店,到手价 ¥123.50",
+        "total_ms": 12345,
     }
 
 
@@ -71,6 +72,8 @@ def test_report_and_derive(client) -> None:
     assert d["information"] == "在美团找到同店,到手价 ¥123.50"
     assert d["store_name"] == "海底捞(朝阳店)"
     assert d["total_dish_count"] == 3
+    assert d["total_ms"] == 12345
+    assert d["raw_payload"]["total_ms"] == 12345
     assert d["skipped_dish_count"] == 1
     assert d["skipped_dish_names"] == ["黑牛肉卷"]
     assert len(d["comparison_results"]) == 3

tools/meituan_playground.py

@@ -52,7 +52,8 @@ def call_raw(path: str, body_obj: dict) -> dict:
     }
     url = f"{settings.MT_CPS_HOST}{path}"
     t0 = time.time()
-    resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
+    # trust_env=False: 美团是国内域名,强制直连绕开本机代理(代理会掐断 TLS 握手,报 SSL EOF)
+    resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC, trust_env=False)
     ms = int((time.time() - t0) * 1000)
     try:
         j = resp.json()