合并：解决广告收益 eCPM 口径测试冲突

## 问题原因

- 新版客户端通过服务端 harvest 落库，但完成路径没有触发 LLM 调用明细与 TOKEN 成本回填。
- 内部共享密钥不一致或 PriceBot 实例切换后，拉取失败只留下空值，后续没有自动补偿。

## 本次改动

- harvest 完成后立即异步回填 LLM 调用、TOKEN 数及成本快照。
- 抽取统一、幂等的成本回填服务，并增加定时补偿 worker。
- 增加 PriceBot 内部鉴权预检、错误日志和多实例查找兜底。
- 仅回填当前价格配置生效后的终态记录，避免用现价误算更早历史数据。
- 补充环境配置、部署说明和单元测试。

## 验证

- 相关测试：36 passed。
- 静态检查通过，diff check 通过。
- 本地页面显示 ¥0.0139，与数据库精确均值 0.013916 的四舍五入结果一致。

## 上线注意

部署时需确保 app-server 与 PriceBot 的 INTERNAL_API_SECRET 完全一致并重启两个服务；worker 启动后会自动补齐符合条件的历史空值。

---------

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

.env.example

@@ -139,6 +139,11 @@ PRICEBOT_COMPARE_TIMEOUT_SEC=60
 # 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
 # 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
 INTERNAL_API_SECRET=
+# 新版比价 harvest 完成后即时回填；以下 worker 再补偿短暂故障期间漏掉的记录。
+LLM_COST_BACKFILL_ENABLED=true
+LLM_COST_BACKFILL_INTERVAL_SEC=300
+LLM_COST_BACKFILL_BATCH_SIZE=100
+LLM_COST_BACKFILL_LOOKBACK_DAYS=30
 
 # ===== CORS =====
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类

app/admin/repositories/ad_revenue.py

@@ -431,6 +431,38 @@ def ad_revenue_report(
         for k, v in type_map.items()
     }
 
+    # 经营看板的两类 eCPM 必须按真实展示的 SDK eCPM 加权，不能用发奖状态修正后的
+    # 收益反推。Draw 包含新 draw 与历史 feed；看视频包含福利与提现视频。
+    category_map: dict[str, dict] = {}
+    for event in events:
+        category = (
+            "draw" if event["ad_type"] in {"draw", "feed"}
+            else "video" if event["ad_type"] in {"reward_video", "withdrawal_video"}
+            else None
+        )
+        if category is None:
+            continue
+        stat = category_map.setdefault(category, {
+            "impressions": 0,
+            "revenue_yuan": 0.0,
+            "ecpm_fen_sum": 0.0,
+        })
+        impressions = int(event["impressions"])
+        stat["impressions"] += impressions
+        stat["revenue_yuan"] += event["revenue_yuan"]
+        stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(event["ecpm"]) * impressions
+    category_stats = {
+        key: {
+            "impressions": value["impressions"],
+            "revenue_yuan": round(value["revenue_yuan"], 6),
+            "ecpm_yuan": round(
+                value["ecpm_fen_sum"] / value["impressions"] / 100.0,
+                6,
+            ) if value["impressions"] else 0.0,
+        }
+        for key, value in category_map.items()
+    }
+
     # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
     # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
     # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
@@ -485,6 +517,7 @@ def ad_revenue_report(
         "daily": daily,
         "hourly": hourly,
         "type_stats": type_stats,
+        "category_stats": category_stats,
         "scene_stats": scene_stats,
         "dau": dau,
         "items": main_rows[offset:offset + limit],

app/admin/repositories/queries.py

@@ -9,7 +9,7 @@ from datetime import date, datetime, time, timedelta, timezone
 from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo
 
-from sqlalchemy import Select, asc, case, desc, func, or_, select
+from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
@@ -45,6 +45,72 @@ _FEED_SCENE_LABEL = {
 }
 
 
+_DEVICE_MARKETING_NAMES = {
+    "23078RKD5C": "Redmi K60 至尊版",
+    "M2012K11AC": "Redmi K40",
+    "PJA110": "一加 Ace 2 Pro",
+    "PPG-AN00": "荣耀 GT Pro",
+    "V2166BA": "vivo Y77e",
+    "V2309A": "vivo X100",
+}
+
+
+def _device_marketing_name(model: str | None) -> str | None:
+    """把线上已知 Build.MODEL 编码转成用户可识别的商品名。"""
+    if not model:
+        return None
+    return _DEVICE_MARKETING_NAMES.get(model.strip().upper())
+
+
+def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
+    """按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
+    candidates = [
+        item for item in feedbacks if item.device_model and item.device_model.strip()
+    ]
+    for item in candidates:
+        item.device_model_name = _device_marketing_name(item.device_model)
+        item.device_manufacturer = None
+        item.rom_version = None
+    if not candidates:
+        return
+
+    ranked = (
+        select(
+            Feedback.id.label("feedback_id"),
+            ComparisonRecord.device_manufacturer.label("device_manufacturer"),
+            ComparisonRecord.rom_version.label("rom_version"),
+            func.row_number()
+            .over(
+                partition_by=Feedback.id,
+                order_by=(
+                    ComparisonRecord.created_at.desc(),
+                    ComparisonRecord.id.desc(),
+                ),
+            )
+            .label("row_num"),
+        )
+        .join(
+            ComparisonRecord,
+            and_(
+                ComparisonRecord.user_id == Feedback.user_id,
+                ComparisonRecord.device_model == Feedback.device_model,
+                ComparisonRecord.created_at <= Feedback.created_at,
+            ),
+        )
+        .where(Feedback.id.in_([item.id for item in candidates]))
+        .subquery()
+    )
+    details = {
+        row.feedback_id: row
+        for row in db.execute(select(ranked).where(ranked.c.row_num == 1)).all()
+    }
+
+    for item in candidates:
+        detail = details.get(item.id)
+        item.device_manufacturer = detail.device_manufacturer if detail else None
+        item.rom_version = detail.rom_version if detail else None
+
+
 def cursor_paginate(
     db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
 ) -> tuple[list, int | None]:
@@ -817,6 +883,7 @@ def list_feedbacks(
         db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
     )
     _attach_user_info(db, items)  # 列表展示完整手机号(点手机号查该用户全部反馈)
+    _attach_feedback_device_details(db, items)
     return items, next_cursor, total
 
 

app/admin/schemas/ad_revenue.py

@@ -75,6 +75,7 @@ class AdRevenueTypeStat(BaseModel):
 
     impressions: int = Field(..., description="该类型展示条数合计")
     revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
+    ecpm_yuan: float | None = Field(None, description="按真实展示次数加权的 SDK eCPM(元/千次)")
 
 
 class AdRevenueRow(BaseModel):
@@ -144,6 +145,10 @@ class AdRevenueReportOut(BaseModel):
         default_factory=dict,
         description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
     )
+    category_stats: dict[str, AdRevenueTypeStat] = Field(
+        default_factory=dict,
+        description="按经营分类小计：draw=draw+历史 feed，video=reward_video+withdrawal_video",
+    )
     scene_stats: dict[str, AdRevenueTypeStat] = Field(
         default_factory=dict,
         description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"

app/admin/schemas/feedback.py

@@ -32,7 +32,10 @@ class FeedbackOut(BaseModel):
     # 提交端环境快照(feedback 表列):提交版本号 / 机型OS版本;改版前的历史反馈为 None
     app_version: str | None = None
     device_model: str | None = None
+    device_model_name: str | None = None
+    device_manufacturer: str | None = None
     rom_name: str | None = None
+    rom_version: int | None = None
     android_version: str | None = None
     # 联表瞬态字段(queries._attach_user_info 挂):列表展示完整手机号,点手机号查该用户全部反馈
     phone: str | None = None

app/api/v1/compare.py

@@ -25,7 +25,7 @@ import uuid
 from typing import Any
 
 import httpx
-from fastapi import APIRouter, HTTPException, Request, status
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import DbSession, OptionalUser
@@ -36,6 +36,7 @@ from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare")
 
@@ -80,7 +81,7 @@ def _harvest_running_blocking(
 def _harvest_done_blocking(
     trace_id: str, user_id: int | None, done_params: dict, business_type: str,
     device_id: str | None, device_info: dict | None, trace_url: str | None,
-) -> None:
+) -> int:
     with SessionLocal() as db:
         rec, newly_success = crud_compare.harvest_done(
             db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -102,6 +103,7 @@ def _harvest_done_blocking(
         # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
         # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
         # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
+        return rec.id
 
 
 def _harvest_abort_blocking(
@@ -243,7 +245,12 @@ async def intent_precoupon_step(
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
+async def price_step(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    user: OptionalUser,
+    db: DbSession,
+) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
@@ -252,12 +259,15 @@ async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dic
     if action.get("command") == "done" and not resp.get("continue", True):
         done_params = action.get("params") or {}
         try:
-            await run_in_threadpool(
+            record_id = await run_in_threadpool(
                 _harvest_done_blocking, trace_id, (user.id if user else None),
                 done_params, "food",
                 meta.get("device_id"), meta.get("device_info"),
                 resp.get("trace_url") or done_params.get("trace_url"),
             )
+            background_tasks.add_task(
+                backfill_comparison_llm_cost, record_id, trace_id
+            )
         except Exception as e:  # noqa: BLE001
             logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/compare_record.py

@@ -16,8 +16,6 @@ import logging
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
 
 from app.api.deps import CurrentUser, DbSession
-from app.db.session import SessionLocal
-from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
@@ -30,8 +28,7 @@ from app.schemas.compare_record import (
     ComparisonRecordOut,
     ComparisonRecordPage,
 )
-from app.services.llm_cost import compute_llm_cost, get_llm_prices
-from app.services.pricebot_llm_calls import fetch_llm_calls
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare_record")
 
@@ -121,32 +118,7 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
     """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
     独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    calls = fetch_llm_calls(trace_id)
-    if not calls:
-        return
-    db = SessionLocal()
-    try:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is None:
-            return
-        rec.llm_calls = calls
-        rec.llm_call_count = len(calls)
-        rec.retry_count = sum(1 for c in calls if c.get("error"))
-        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
-        # error 的调用 usage 可能为 None,or {} 兜底)
-        rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
-        rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
-        # 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
-        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
-        db.commit()
-        logger.info(
-            "backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
-            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
-        )
-    except Exception as e:  # noqa: BLE001 best-effort
-        logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
-    finally:
-        db.close()
+    backfill_comparison_llm_cost(record_id, trace_id)
 
 
 @router.get(

app/core/config.py

@@ -377,6 +377,12 @@ class Settings(BaseSettings):
     # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
     # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
     INTERNAL_API_SECRET: str = ""
+    # Current compare clients are persisted by server-side harvest. Repair any
+    # recent terminal rows left without LLM usage by transient upstream/auth failures.
+    LLM_COST_BACKFILL_ENABLED: bool = True
+    LLM_COST_BACKFILL_INTERVAL_SEC: int = 300
+    LLM_COST_BACKFILL_BATCH_SIZE: int = 100
+    LLM_COST_BACKFILL_LOOKBACK_DAYS: int = 30
 
     # ===== 媒体文件(用户头像上传)=====
     # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。

app/core/llm_cost_backfill_worker.py

@@ -0,0 +1,63 @@
+"""Periodic repair worker for comparison records with missing LLM token cost."""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+from app.core.config import settings
+from app.services.comparison_llm_backfill import repair_missing_comparison_llm_costs
+from app.services.pricebot_llm_calls import pricebot_llm_auth_ready
+
+logger = logging.getLogger("shagua.llm_cost_backfill_worker")
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.LLM_COST_BACKFILL_INTERVAL_SEC))
+    logger.info(
+        "LLM cost backfill worker started interval=%ss batch=%s lookback_days=%s",
+        interval,
+        settings.LLM_COST_BACKFILL_BATCH_SIZE,
+        settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
+    )
+    try:
+        while True:
+            try:
+                auth_ready = await asyncio.to_thread(pricebot_llm_auth_ready)
+                if auth_ready:
+                    result = await asyncio.to_thread(
+                        repair_missing_comparison_llm_costs,
+                        limit=settings.LLM_COST_BACKFILL_BATCH_SIZE,
+                        lookback_days=settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
+                    )
+                    logger.info("LLM cost backfill batch result=%s", result)
+                else:
+                    logger.error(
+                        "LLM cost backfill skipped: PriceBot internal auth is not ready"
+                    )
+            except Exception:  # noqa: BLE001
+                logger.exception("LLM cost backfill batch failed")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("LLM cost backfill worker stopped")
+        raise
+
+
+def start_llm_cost_backfill_worker() -> asyncio.Task | None:
+    if not settings.LLM_COST_BACKFILL_ENABLED:
+        logger.info("LLM cost backfill worker disabled")
+        return None
+    if not settings.INTERNAL_API_SECRET:
+        logger.warning(
+            "LLM cost backfill worker not started: INTERNAL_API_SECRET is empty"
+        )
+        return None
+    return asyncio.create_task(_run_loop(), name="llm-cost-backfill")
+
+
+async def stop_llm_cost_backfill_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/main.py

@@ -60,6 +60,10 @@ from app.core.inactivity_reset_worker import (
     start_inactivity_reset_worker,
     stop_inactivity_reset_worker,
 )
+from app.core.llm_cost_backfill_worker import (
+    start_llm_cost_backfill_worker,
+    stop_llm_cost_backfill_worker,
+)
 from app.core.logging import setup_logging
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
@@ -103,6 +107,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
     inactivity_task = start_inactivity_reset_worker()
+    llm_cost_backfill_task = start_llm_cost_backfill_worker()
     try:
         yield
     finally:
@@ -112,6 +117,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
+        await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
         await aclose_pricebot_client()
         mt_meituan.close_client()
         logger.info("shutting down")

app/services/comparison_llm_backfill.py

@@ -0,0 +1,156 @@
+"""Persist and repair comparison-record LLM token costs."""
+from __future__ import annotations
+
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services.llm_cost import compute_llm_cost, get_llm_prices
+from app.services.pricebot_llm_calls import fetch_llm_calls
+
+logger = logging.getLogger("shagua.comparison_llm_backfill")
+
+
+def _utc_to_beijing_naive(value: datetime) -> datetime:
+    """Convert a DB UTC timestamp to comparison_record's Beijing wall-clock."""
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=UTC)
+    return value.astimezone(CN_TZ).replace(tzinfo=None)
+
+
+def _store_calls(record_id: int, trace_id: str, calls: list[dict]) -> bool:
+    """Store calls and all derived fields atomically."""
+    with SessionLocal() as db:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is None or rec.trace_id != trace_id:
+            logger.warning(
+                "LLM cost backfill record mismatch record_id=%s trace=%s",
+                record_id,
+                trace_id,
+            )
+            return False
+
+        # Never recalculate a frozen historical cost with a newer price config.
+        if rec.llm_cost_yuan is not None and rec.llm_calls:
+            return False
+
+        rec.llm_calls = calls
+        rec.llm_call_count = len(calls)
+        rec.retry_count = sum(1 for call in calls if call.get("error"))
+        rec.input_tokens = sum(
+            (call.get("usage") or {}).get("prompt_tokens") or 0 for call in calls
+        )
+        rec.output_tokens = sum(
+            (call.get("usage") or {}).get("completion_tokens") or 0 for call in calls
+        )
+        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(
+            calls, get_llm_prices(db)
+        )
+        db.commit()
+        logger.info(
+            "LLM cost backfilled trace=%s calls=%d input_tokens=%d "
+            "output_tokens=%d cost=%s",
+            trace_id,
+            len(calls),
+            rec.input_tokens,
+            rec.output_tokens,
+            rec.llm_cost_yuan,
+        )
+        return True
+
+
+def backfill_comparison_llm_cost(
+    record_id: int,
+    trace_id: str,
+    *,
+    attempts: int = 3,
+    retry_delays: tuple[float, ...] = (1.0, 3.0),
+) -> bool:
+    """Fetch and persist one record, retrying short-lived upstream races."""
+    if not settings.INTERNAL_API_SECRET or not trace_id:
+        logger.warning(
+            "LLM cost backfill skipped trace=%s: INTERNAL_API_SECRET is not configured",
+            trace_id,
+        )
+        return False
+    total_attempts = max(1, attempts)
+    for attempt in range(total_attempts):
+        calls = fetch_llm_calls(trace_id)
+        if calls:
+            try:
+                return _store_calls(record_id, trace_id, calls)
+            except Exception:  # noqa: BLE001 - background repair must stay alive
+                logger.exception(
+                    "LLM cost store failed trace=%s record_id=%s",
+                    trace_id,
+                    record_id,
+                )
+                return False
+
+        if attempt + 1 < total_attempts:
+            delay = retry_delays[min(attempt, len(retry_delays) - 1)] if retry_delays else 0
+            if delay > 0:
+                time.sleep(delay)
+
+    logger.warning(
+        "LLM cost backfill has no calls trace=%s record_id=%s attempts=%d",
+        trace_id,
+        record_id,
+        total_attempts,
+    )
+    return False
+
+
+def repair_missing_comparison_llm_costs(
+    *,
+    limit: int = 100,
+    lookback_days: int = 30,
+) -> dict[str, int]:
+    """Repair a bounded batch of recent terminal records with missing cost."""
+    cutoff = datetime.now(CN_TZ).replace(tzinfo=None) - timedelta(
+        days=max(1, lookback_days)
+    )
+    with SessionLocal() as db:
+        # app_config has no price history. Repricing a record from before the
+        # current config became effective would fabricate a historical cost, so
+        # only repair records at/after that timestamp.
+        price_config_updated_at = db.execute(
+            select(AppConfig.updated_at).where(AppConfig.key == "llm_token_price")
+        ).scalar_one_or_none()
+        date_conditions = [ComparisonRecord.created_at >= cutoff]
+        if price_config_updated_at is not None:
+            date_conditions.append(
+                ComparisonRecord.created_at
+                >= _utc_to_beijing_naive(price_config_updated_at)
+            )
+        candidates = list(
+            db.execute(
+                select(ComparisonRecord.id, ComparisonRecord.trace_id)
+                .where(
+                    *date_conditions,
+                    ComparisonRecord.status.in_(("success", "failed")),
+                    ComparisonRecord.llm_cost_yuan.is_(None),
+                )
+                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
+                .limit(max(1, limit))
+            ).all()
+        )
+
+    repaired = 0
+    for record_id, trace_id in candidates:
+        if backfill_comparison_llm_cost(
+            record_id, trace_id, attempts=1, retry_delays=()
+        ):
+            repaired += 1
+    return {
+        "candidates": len(candidates),
+        "repaired": repaired,
+        "unresolved": len(candidates) - repaired,
+    }

app/services/pricebot_llm_calls.py

@@ -20,19 +20,67 @@ from app.core.pricebot_router import pick_pricebot
 logger = logging.getLogger("shagua.pricebot_llm")
 
 
+def pricebot_llm_auth_ready() -> bool:
+    """Verify every configured PriceBot instance accepts the shared secret."""
+    secret = settings.INTERNAL_API_SECRET
+    if not secret:
+        logger.error("PriceBot LLM auth check failed: INTERNAL_API_SECRET is empty")
+        return False
+    for base in settings.pricebot_instances:
+        url = f"{base.rstrip('/')}/api/internal/llm_calls/__auth_probe__"
+        try:
+            resp = httpx.get(
+                url, headers={"X-Internal-Secret": secret}, timeout=3.0
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.error("PriceBot LLM auth check unavailable base=%s: %s", base, exc)
+            return False
+        if resp.status_code != 200:
+            logger.error(
+                "PriceBot LLM auth check rejected base=%s status=%s; "
+                "verify both services use the same INTERNAL_API_SECRET",
+                base,
+                resp.status_code,
+            )
+            return False
+    return True
+
+
 def fetch_llm_calls(trace_id: str) -> list[dict]:
     """返回该次比价的 LLM 调用明细列表(每条 {scene,model,input_messages,output,usage,latency_ms,error});
     未配密钥 / 无 trace_id / 拉取失败 → []。"""
     secret = settings.INTERNAL_API_SECRET
     if not secret or not trace_id:
         return []
-    base = pick_pricebot(trace_id).rstrip("/")
-    url = f"{base}/api/internal/llm_calls/{trace_id}"
-    try:
-        resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
-        if resp.status_code == 200:
-            return resp.json().get("calls", []) or []
-        logger.warning("fetch_llm_calls trace=%s status=%s", trace_id, resp.status_code)
-    except Exception as e:  # noqa: BLE001 — best-effort,任何异常都不该影响上报
-        logger.warning("fetch_llm_calls trace=%s failed: %s", trace_id, e)
+    preferred = pick_pricebot(trace_id)
+    # LLM JSONL is instance-local. If the cluster topology changed after a
+    # historical trace was created, consistent hashing may now point elsewhere;
+    # probe the remaining configured instances only when the preferred one is empty.
+    bases = [preferred, *(base for base in settings.pricebot_instances if base != preferred)]
+    for base in bases:
+        url = f"{base.rstrip('/')}/api/internal/llm_calls/{trace_id}"
+        try:
+            resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
+            if resp.status_code == 200:
+                calls = resp.json().get("calls", []) or []
+                if calls:
+                    return calls
+                continue
+            if resp.status_code in (401, 403):
+                logger.error(
+                    "fetch_llm_calls rejected trace=%s base=%s status=%s; "
+                    "INTERNAL_API_SECRET differs between app-server and PriceBot",
+                    trace_id,
+                    base,
+                    resp.status_code,
+                )
+            else:
+                logger.warning(
+                    "fetch_llm_calls trace=%s base=%s status=%s",
+                    trace_id,
+                    base,
+                    resp.status_code,
+                )
+        except Exception as e:  # noqa: BLE001 — best-effort
+            logger.warning("fetch_llm_calls trace=%s base=%s failed: %s", trace_id, base, e)
     return []

deploy/comparison-llm-cost.md

@@ -0,0 +1,40 @@
+# 比价 TOKEN 成本采集与补偿
+
+## 部署前置
+
+App Server 与 PriceBot 使用各自独立的 `.env`，但下面的值必须完全一致：
+
+- `/opt/shaguabijia-app-server/.env`
+- `/opt/pricebot-backend/.env`
+- 配置项：`INTERNAL_API_SECRET`
+
+不要把密钥原文写入日志、命令历史或 Git。修改后同时重启两个服务。
+
+App Server 启动后会逐个探测 `PRICEBOT_INSTANCES` 的内部读取接口。鉴权不一致时会记录
+`PriceBot LLM auth check rejected`，并跳过本轮补偿，避免对所有缺失记录重复发送失败请求。
+
+## 数据链路
+
+1. 当前客户端由 App Server 在 PriceBot 最终 `done` 帧到达时 harvest 比价记录。
+2. harvest 成功后立即异步读取同一 `trace_id` 的 LLM 调用，冻结 Token、成本和单价快照。
+3. 周期 worker 扫描近期 `success/failed` 且 `llm_cost_yuan IS NULL` 的记录进行补偿；
+   为避免用现价伪造历史成本，只处理当前单价配置生效时间之后的记录。
+4. 管理后台顶部“平均 TOKEN 成本”使用筛选范围内已冻结成本的数据库平均值。
+
+## 上线验收（只读 SQL）
+
+```sql
+SELECT
+  (created_at AT TIME ZONE 'Asia/Shanghai')::date AS day,
+  count(*) AS records,
+  count(llm_cost_yuan) AS cost_records,
+  round(avg(llm_cost_yuan)::numeric, 6) AS avg_token_cost
+FROM comparison_record
+WHERE created_at >= now() - interval '3 days'
+GROUP BY 1
+ORDER BY 1 DESC;
+```
+
+新产生的正常终态比价记录应在短时间内写入 `input_tokens`、`output_tokens` 和
+`llm_cost_yuan`。历史记录只有在 PriceBot 的对应 trace JSONL 仍保留时才能准确回填；
+原始调用已经清理的记录不能用估算值冒充真实成本。

scripts/seed_mock_feedback.py

@@ -175,22 +175,22 @@ def seed(db) -> list[Feedback]:
         # 普通反馈 · 新端(带环境快照)· 无图
         fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
            source="profile",
-           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
+           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS 14", android_version="14",
            created=ago(minutes=6)),
         # 比价反馈 · scene=优惠不对 · 新端 · 2 图
         fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
            source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
-           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
+           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
            created=ago(minutes=22)),
         # 比价反馈 · scene=找错商品 · 新端 · 无图
         fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
            source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
+           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", android_version="14",
            created=ago(hours=1)),
         # 普通反馈 · 新端 · 1 图(表扬 + 小问题)
         fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
            source="profile", images=[imgs[2]],
-           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
+           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI 14", android_version="14",
            created=ago(hours=3)),
         # 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
         fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
@@ -206,7 +206,7 @@ def seed(db) -> list[Feedback]:
            source="profile", images=[imgs[3]],
            status="adopted", reward_coins=2000,
            review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
-           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
+           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS 13", android_version="13",
            created=ago(days=2)),
 
         # ===== 未采纳 rejected(带原因 + 回复)=====
@@ -214,7 +214,7 @@ def seed(db) -> list[Feedback]:
            source="comparison", scene="价格不准",
            status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
            admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
-           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
+           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
            created=ago(days=3)),
     ]
     db.add_all(feedbacks)

tests/test_admin_ad_revenue_scope.py

@@ -211,6 +211,59 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         db.close()
 
 
+def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
+    db = SessionLocal()
+    phone = "18800009993"
+    category_date = "2040-02-05"
+    try:
+        user = User(phone=phone, username="29999999993", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="category-draw",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="feed", ad_session_id="category-feed",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=category_date,
+            date_to=category_date,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+
+        assert result["category_stats"] == {
+            "draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
+            "video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
 def test_feed_reward_details_keep_each_record_adn() -> None:
     db = SessionLocal()
     phone = "18800009994"

tests/test_admin_feedback_device_details.py

@@ -0,0 +1,71 @@
+from datetime import datetime, timedelta, timezone
+
+from app.admin.repositories import queries
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
+from app.models.feedback import Feedback
+from app.repositories import user as user_repo
+
+
+def test_feedback_device_details_use_same_user_and_model() -> None:
+    with SessionLocal() as db:
+        submitted_at = datetime.now(timezone.utc)
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone="13800009876",
+            register_channel="sms",
+        )
+        db.add_all(
+            [
+                ComparisonRecord(
+                    user_id=user.id,
+                    trace_id="feedback-device-v2166ba",
+                    device_model="V2166BA",
+                    device_manufacturer="vivo",
+                    rom_name="OriginOS",
+                    rom_version=13,
+                    created_at=submitted_at - timedelta(minutes=5),
+                ),
+                ComparisonRecord(
+                    user_id=user.id,
+                    trace_id="feedback-device-other",
+                    device_model="OTHER-CODE",
+                    device_manufacturer="Other",
+                    rom_name="OtherOS",
+                    rom_version=99,
+                    created_at=submitted_at - timedelta(minutes=5),
+                ),
+                ComparisonRecord(
+                    user_id=user.id,
+                    trace_id="feedback-device-future-upgrade",
+                    device_model="V2166BA",
+                    device_manufacturer="vivo-new",
+                    rom_name="OriginOS",
+                    rom_version=99,
+                    created_at=submitted_at + timedelta(minutes=5),
+                ),
+                Feedback(
+                    user_id=user.id,
+                    content="设备信息补全测试",
+                    contact="",
+                    status="pending",
+                    device_model="V2166BA",
+                    rom_name="OriginOS",
+                    android_version="13",
+                    created_at=submitted_at,
+                ),
+            ]
+        )
+        db.commit()
+
+        items, _next_cursor, total = queries.list_feedbacks(
+            db,
+            user_id=user.id,
+            limit=20,
+        )
+
+        assert total == 1
+        assert items[0].device_model == "V2166BA"
+        assert items[0].device_model_name == "vivo Y77e"
+        assert items[0].device_manufacturer == "vivo"
+        assert items[0].rom_version == 13

tests/test_compare_harvest.py

@@ -9,17 +9,16 @@ pricebot 用 httpx mock,不真连(同 test_compare_proxy)。
 """
 from __future__ import annotations
 
-import json
 import uuid
 from unittest.mock import MagicMock, patch
 
 import httpx
+from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
 from app.schemas.compare_record import ComparisonRecordIn
-from sqlalchemy import select
 
 
 def _tid() -> str:
@@ -216,7 +215,9 @@ def test_price_step_done_harvests_success(client) -> None:
                   "action": {"command": "done", "params": _done_params()},
                   "trace_url": "https://price.shaguabijia.com/traces/done2/"}
     p, _cap = _mock_pricebot(done_frame)
-    with p:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
         r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
     assert r.status_code == 200
     with SessionLocal() as db:
@@ -224,6 +225,7 @@ def test_price_step_done_harvests_success(client) -> None:
         assert rec is not None and rec.status == "success"
         assert rec.best_platform_id == "meituan"
         assert rec.saved_amount_cents == 500
+        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_trace_finalize_harvests_abort(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services import comparison_llm_backfill
+
+
+def _record(
+    trace_id: str,
+    *,
+    status: str = "success",
+    created_at: datetime | None = None,
+) -> int:
+    with SessionLocal() as db:
+        rec = ComparisonRecord(
+            trace_id=trace_id,
+            status=status,
+            created_at=created_at or datetime.now(),
+        )
+        db.add(rec)
+        db.commit()
+        return rec.id
+
+
+def _delete(record_id: int) -> None:
+    with SessionLocal() as db:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is not None:
+            db.delete(rec)
+            db.commit()
+
+
+def test_backfill_retries_then_persists_cost(monkeypatch):
+    record_id = _record("llm-retry-1")
+    calls = [
+        {
+            "model": "unknown-model",
+            "error": None,
+            "usage": {"prompt_tokens": 1000, "completion_tokens": 500},
+        }
+    ]
+    responses = iter([[], calls])
+    monkeypatch.setattr(
+        comparison_llm_backfill,
+        "fetch_llm_calls",
+        lambda trace_id: next(responses),
+    )
+    sleeps: list[float] = []
+    monkeypatch.setattr(comparison_llm_backfill.time, "sleep", sleeps.append)
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
+
+    try:
+        assert comparison_llm_backfill.backfill_comparison_llm_cost(
+            record_id, "llm-retry-1", attempts=2, retry_delays=(0.25,)
+        )
+        assert sleeps == [0.25]
+        with SessionLocal() as db:
+            rec = db.get(ComparisonRecord, record_id)
+            assert rec.input_tokens == 1000
+            assert rec.output_tokens == 500
+            assert rec.llm_cost_yuan is not None
+            assert rec.llm_calls == calls
+    finally:
+        _delete(record_id)
+
+
+def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
+    missing_id = _record("llm-repair-missing")
+    running_id = _record("llm-repair-running", status="running")
+    calls = [
+        {
+            "model": "unknown-model",
+            "error": None,
+            "usage": {"prompt_tokens": 100, "completion_tokens": 20},
+        }
+    ]
+    seen: list[str] = []
+
+    def fetch(trace_id: str) -> list[dict]:
+        seen.append(trace_id)
+        return calls
+
+    monkeypatch.setattr(comparison_llm_backfill, "fetch_llm_calls", fetch)
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
+    try:
+        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
+            limit=10, lookback_days=1
+        )
+        assert result["repaired"] >= 1
+        assert "llm-repair-missing" in seen
+        assert "llm-repair-running" not in seen
+        with SessionLocal() as db:
+            assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
+            assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
+    finally:
+        _delete(missing_id)
+        _delete(running_id)
+
+
+def test_repair_excludes_records_before_current_price_config(monkeypatch):
+    price_changed_at = datetime.now(UTC) - timedelta(hours=1)
+    before_change = (price_changed_at - timedelta(minutes=30)).astimezone(CN_TZ)
+    after_change = (price_changed_at + timedelta(minutes=30)).astimezone(CN_TZ)
+    before_id = _record(
+        "llm-before-price-change",
+        created_at=before_change.replace(tzinfo=None),
+    )
+    after_id = _record(
+        "llm-after-price-change",
+        created_at=after_change.replace(tzinfo=None),
+    )
+    with SessionLocal() as db:
+        existing = db.get(AppConfig, "llm_token_price")
+        if existing is not None:
+            db.delete(existing)
+            db.flush()
+        db.add(
+            AppConfig(
+                key="llm_token_price",
+                value={"default": {"input_per_1m": 1, "output_per_1m": 1}},
+                updated_at=price_changed_at,
+            )
+        )
+        db.commit()
+
+    seen: list[str] = []
+
+    def backfill(record_id: int, trace_id: str, **kwargs) -> bool:
+        seen.append(trace_id)
+        return True
+
+    monkeypatch.setattr(
+        comparison_llm_backfill,
+        "backfill_comparison_llm_cost",
+        backfill,
+    )
+    try:
+        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
+            limit=10_000,
+            lookback_days=1,
+        )
+        assert "llm-after-price-change" in seen
+        assert "llm-before-price-change" not in seen
+        assert result["repaired"] == len(seen)
+        assert result["unresolved"] == 0
+    finally:
+        _delete(before_id)
+        _delete(after_id)
+        with SessionLocal() as db:
+            config = db.get(AppConfig, "llm_token_price")
+            if config is not None:
+                db.delete(config)
+                db.commit()

tests/test_llm_cost.py

@@ -132,6 +132,7 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
     from app.models.app_config import AppConfig
     from app.models.comparison import ComparisonRecord
     from app.repositories import app_config
+    from app.services import comparison_llm_backfill
 
     sample = [
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
@@ -139,7 +140,12 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
     ]
-    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
+    monkeypatch.setattr(
+        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
+    )
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
 
     db = SessionLocal()
     try:

tests/test_pricebot_llm_calls.py

@@ -0,0 +1,55 @@
+from unittest.mock import MagicMock
+
+from app.core.config import settings
+from app.services import pricebot_llm_calls
+
+
+def test_auth_probe_accepts_matching_secret(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
+    response = MagicMock(status_code=200)
+    get = MagicMock(return_value=response)
+    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
+
+    assert pricebot_llm_calls.pricebot_llm_auth_ready() is True
+    assert get.call_count == len(settings.pricebot_instances)
+    assert all(
+        call.kwargs["headers"]["X-Internal-Secret"] == "same-secret"
+        for call in get.call_args_list
+    )
+
+
+def test_auth_probe_rejects_mismatched_secret(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "app-server-secret")
+    monkeypatch.setattr(
+        pricebot_llm_calls.httpx,
+        "get",
+        MagicMock(return_value=MagicMock(status_code=403)),
+    )
+
+    assert pricebot_llm_calls.pricebot_llm_auth_ready() is False
+
+
+def test_fetch_falls_back_to_other_instance_when_hash_target_is_empty(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
+    monkeypatch.setattr(
+        settings,
+        "PRICEBOT_INSTANCES",
+        "http://pricebot-1:8000,http://pricebot-2:8000",
+    )
+    monkeypatch.setattr(
+        pricebot_llm_calls,
+        "pick_pricebot",
+        lambda trace_id: "http://pricebot-1:8000",
+    )
+    calls = [{"model": "qwen", "usage": {"prompt_tokens": 1}}]
+
+    def get(url, **kwargs):
+        response = MagicMock(status_code=200)
+        response.json.return_value = {
+            "calls": [] if "pricebot-1" in url else calls
+        }
+        return response
+
+    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
+
+    assert pricebot_llm_calls.fetch_llm_calls("trace-after-rescale") == calls