修复：补全反馈设备与系统版本

反馈列表按同一用户、同一设备和提交时间补全可读机型及厂商系统版本，并更新 mock 数据与回归测试。

.env.example

@@ -139,11 +139,6 @@ 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_audit.py

@@ -12,11 +12,10 @@
 """
 from __future__ import annotations
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
-from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.repositories.ad_feed_reward import FEED_REWARD_UNIT_SECONDS
@@ -56,21 +55,10 @@ def _reward_video_rows(
     if user_id is not None:
         stmt = stmt.where(AdRewardRecord.user_id == user_id)
 
-    records = list(db.execute(stmt).scalars())
-    # S2S 发奖回调不携带实际填充 ADN；用相同用户和 ad_session_id 的展示记录回填。
-    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
-    impression_by_session = {
-        (record.user_id, record.ad_session_id): record
-        for record in db.execute(
-            select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
-        ).scalars()
-    } if session_ids else {}
-
     # 用本日之前的累计份数做起点,当日 granted 在其上继续递增 → 与 _granted_cumulative+1 对齐
     granted_n: dict[int, int] = _prior_granted_counts(db, date=date, user_id=user_id)
     rows: list[dict] = []
-    for rec in records:
-        impression = impression_by_session.get((rec.user_id, rec.ad_session_id))
+    for rec in db.execute(stmt).scalars():
         if rec.status == "granted":
             nth = granted_n.get(rec.user_id, 0) + 1
             granted_n[rec.user_id] = nth
@@ -80,8 +68,6 @@ def _reward_video_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
-                "adn": impression.adn if impression is not None else None,
-                "slot_id": impression.slot_id if impression is not None else None,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -104,8 +90,6 @@ def _reward_video_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
-                "adn": impression.adn if impression is not None else None,
-                "slot_id": impression.slot_id if impression is not None else None,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -165,78 +149,6 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
     return True
 
 
-def _nonblank(value: str | None) -> str | None:
-    value = value.strip() if value else None
-    return value or None
-
-
-def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | None]:
-    """仅在候选展示记录指向唯一 ADN 时回填来源，避免错误归因。"""
-    adns = {_nonblank(record.adn) for record in records}
-    adns.discard(None)
-    if len(adns) != 1:
-        return None, None
-
-    slots = {_nonblank(record.slot_id) for record in records}
-    slots.discard(None)
-    return next(iter(adns)), next(iter(slots)) if len(slots) == 1 else None
-
-
-def _feed_source_fallbacks(
-    db: Session, records: list[AdFeedRewardRecord]
-) -> tuple[
-    dict[tuple[int, str], tuple[str | None, str | None]],
-    dict[tuple[int, str, str], tuple[str | None, str | None]],
-]:
-    """为旧信息流发奖记录构建安全来源索引。"""
-    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
-    trace_ids = {record.trace_id for record in records if record.trace_id}
-    if not session_ids and not trace_ids:
-        return {}, {}
-
-    filters = []
-    if session_ids:
-        filters.append(AdEcpmRecord.ad_session_id.in_(session_ids))
-    if trace_ids:
-        filters.append(AdEcpmRecord.trace_id.in_(trace_ids))
-    impressions = list(db.execute(select(AdEcpmRecord).where(or_(*filters))).scalars())
-
-    by_session: dict[tuple[int, str], list[AdEcpmRecord]] = {}
-    by_trace_ecpm: dict[tuple[int, str, str], list[AdEcpmRecord]] = {}
-    for impression in impressions:
-        if impression.ad_session_id:
-            by_session.setdefault((impression.user_id, impression.ad_session_id), []).append(impression)
-        if impression.trace_id:
-            by_trace_ecpm.setdefault(
-                (impression.user_id, impression.trace_id, impression.ecpm_raw), []
-            ).append(impression)
-
-    return (
-        {key: _unique_ad_source(value) for key, value in by_session.items()},
-        {key: _unique_ad_source(value) for key, value in by_trace_ecpm.items()},
-    )
-
-
-def _feed_source(
-    record: AdFeedRewardRecord,
-    *,
-    by_session: dict[tuple[int, str], tuple[str | None, str | None]],
-    by_trace_ecpm: dict[tuple[int, str, str], tuple[str | None, str | None]],
-) -> tuple[str | None, str | None]:
-    """返回本条发奖广告的来源；无唯一证据时保留原始空值。"""
-    adn, slot_id = _nonblank(record.adn), _nonblank(record.slot_id)
-    if adn and slot_id:
-        return adn, slot_id
-
-    candidate = by_session.get((record.user_id, record.ad_session_id or ""))
-    if candidate is None and record.trace_id:
-        candidate = by_trace_ecpm.get((record.user_id, record.trace_id, record.ecpm_raw))
-    if candidate is None:
-        return adn, slot_id
-    candidate_adn, candidate_slot_id = candidate
-    return adn or candidate_adn, slot_id or candidate_slot_id
-
-
 def _feed_rows(
     db: Session, *, date: str, user_id: int | None, scene: str | None = None
 ) -> list[dict]:
@@ -255,17 +167,11 @@ def _feed_rows(
     if user_id is not None:
         stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
 
-    records = list(db.execute(stmt).scalars())
-    by_session, by_trace_ecpm = _feed_source_fallbacks(db, records)
-
     # 本日之前的累计**条数**做起点,与发奖侧 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 records:
+    for rec in db.execute(stmt).scalars():
         keep = _feed_scene_matches(rec, scene)  # 累计照常推进,这里只决定是否展示本行
-        adn, slot_id = _feed_source(
-            rec, by_session=by_session, by_trace_ecpm=by_trace_ecpm
-        )
         if rec.status == "granted":
             # 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
             # nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
@@ -282,8 +188,6 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
-                "adn": adn,
-                "slot_id": slot_id,
                 "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
@@ -310,8 +214,6 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
-                "adn": adn,
-                "slot_id": slot_id,
                 "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,

app/admin/repositories/ad_revenue.py

@@ -96,11 +96,7 @@ _REWARD_DETAIL_KEYS = (
 
 def _reward_detail(row: dict) -> dict:
     """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
-    detail = {key: row[key] for key in _REWARD_DETAIL_KEYS}
-    # 聚合父行可能包含多个 ADN，来源必须保留在每一条发奖明细上。
-    detail["adn"] = row.get("adn")
-    detail["slot_id"] = row.get("slot_id")
-    return detail
+    return {k: row[k] for k in _REWARD_DETAIL_KEYS}
 
 
 def ad_revenue_report(
@@ -248,8 +244,8 @@ def ad_revenue_report(
                 "impressions": 0,
                 "ecpm": row["ecpm"],
                 "revenue_yuan": 0.0,
-                "adn": row.get("adn"),
-                "slot_id": row.get("slot_id"),
+                "adn": None,
+                "slot_id": None,
                 "has_reward": True,
                 "status": row["status"],
                 "expected_coin": int(row["expected_coin"]),

app/admin/repositories/queries.py

@@ -1295,11 +1295,6 @@ def _cn_wall_to_utc(dt: datetime) -> datetime:
     return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
 
 
-def _coin_record_sort_key(row: dict) -> datetime:
-    """金币明细跨数据源排序键：兼容 SQLite naive 与 PostgreSQL aware 时间。"""
-    return _as_utc(row["created_at"])
-
-
 def user_coin_records(
     db: Session,
     user_id: int,
@@ -1391,10 +1386,7 @@ def user_coin_records(
             "coin": rec.amount,
         })
 
-    # SQLite 常返回 naive datetime，PostgreSQL timestamptz 返回 aware datetime；
-    # 统一成 aware UTC 排序，避免线上合并广告记录与签到记录时抛
-    # “can't compare offset-naive and offset-aware datetimes”。
-    rows.sort(key=_coin_record_sort_key, reverse=True)
+    rows.sort(key=lambda r: r["created_at"], reverse=True)
     has_more = len(rows) > offset + limit
 
     # 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)

app/admin/routers/users.py

@@ -86,7 +86,7 @@ def get_user_reward_stats(
     date_to: Annotated[datetime | None, Query()] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
-    if not user_repo.user_exists(db, user_id):
+    if user_repo.get_user_by_id(db, user_id) is None:
         raise HTTPException(status_code=404, detail="用户不存在")
     return UserRewardStats(
         **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)

app/admin/schemas/ad_revenue.py

@@ -40,8 +40,6 @@ class AdRevenueRecord(BaseModel):
     expected_coin: int = Field(..., description="按公式复算应发金币")
     actual_coin: int = Field(..., description="实际入账金币")
     matched: bool = Field(..., description="复算与实发是否一致")
-    adn: str | None = Field(None, description="本条发奖对应的实际填充 ADN 子渠道")
-    slot_id: str | None = Field(None, description="本条发奖对应的底层 mediation rit")
 
 
 class AdRevenueDaily(BaseModel):

app/api/v1/compare.py

@@ -25,7 +25,7 @@ import uuid
 from typing import Any
 
 import httpx
-from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
+from fastapi import APIRouter, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import DbSession, OptionalUser
@@ -36,7 +36,6 @@ 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")
 
@@ -81,7 +80,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,
-) -> int:
+) -> None:
     with SessionLocal() as db:
         rec, newly_success = crud_compare.harvest_done(
             db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -103,7 +102,6 @@ def _harvest_done_blocking(
         # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
         # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
         # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
-        return rec.id
 
 
 def _harvest_abort_blocking(
@@ -245,12 +243,7 @@ async def intent_precoupon_step(
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(
-    request: Request,
-    background_tasks: BackgroundTasks,
-    user: OptionalUser,
-    db: DbSession,
-) -> dict[str, Any]:
+async def price_step(request: Request, 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 更新成终态。
@@ -259,15 +252,12 @@ async def price_step(
     if action.get("command") == "done" and not resp.get("continue", True):
         done_params = action.get("params") or {}
         try:
-            record_id = await run_in_threadpool(
+            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,6 +16,8 @@ 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 (
@@ -28,7 +30,8 @@ from app.schemas.compare_record import (
     ComparisonRecordOut,
     ComparisonRecordPage,
 )
-from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
+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.compare_record")
 
@@ -118,7 +121,32 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
     """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
     独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    backfill_comparison_llm_cost(record_id, trace_id)
+    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()
 
 
 @router.get(

app/core/config.py

@@ -377,12 +377,6 @@ 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

@@ -1,63 +0,0 @@
-"""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,10 +60,6 @@ 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 (
@@ -107,7 +103,6 @@ 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:
@@ -117,7 +112,6 @@ 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/repositories/comparison.py

@@ -55,22 +55,13 @@ def _product_names_from_items(items: list | None) -> str | None:
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
-    _pr = payload.platform_results or {}
 
-    def _is_short(r) -> bool:
-        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
-        # platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
-        info = _pr.get(r.platform_id) if r.platform_id else None
-        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
-
-    # 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
-    # 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
-    priced = [r for r in results if r.price is not None]
-    clean = [r for r in priced if not _is_short(r)]
+    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
     best = None
-    if clean:
+    priced = [r for r in results if r.price is not None]
+    if priced:
         best = min(
-            clean,
+            priced,
             key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
         )
 
@@ -205,29 +196,14 @@ def upsert_record(
 # ============================================================
 
 
-def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None
-) -> dict:
+def _derive_from_results(results: list[dict]) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
-
-    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
-    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
-    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
-    _pr = platform_results or {}
-
-    def _is_short(r: dict) -> bool:
-        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
-        pid = r.get("platform_id")
-        info = _pr.get(pid) if pid else None
-        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
-
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
     priced = [r for r in results if r.get("price") is not None]
-    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
     best = None
-    if clean:
+    if priced:
         best = min(
-            clean,
+            priced,
             key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
         )
     src_row = next((r for r in results if r.get("is_source")), None)
@@ -413,7 +389,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
+    derived = _derive_from_results(results)
     # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
     items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(

app/repositories/user.py

@@ -93,11 +93,6 @@ def get_user_by_id(db: Session, user_id: int) -> User | None:
     return db.get(User, user_id)
 
 
-def user_exists(db: Session, user_id: int) -> bool:
-    """只查主键判断用户是否存在，避免只读统计接口依赖完整用户表结构。"""
-    return db.scalar(select(User.id).where(User.id == user_id)) is not None
-
-
 def get_user_by_phone(db: Session, phone: str) -> User | None:
     stmt = select(User).where(User.phone == phone)
     return db.execute(stmt).scalar_one_or_none()

app/services/comparison_llm_backfill.py

@@ -1,156 +0,0 @@
-"""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,67 +20,19 @@ 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 []
-    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)
+    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)
     return []

deploy/comparison-llm-cost.md

@@ -1,40 +0,0 @@
-# 比价 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 仍保留时才能准确回填；
-原始调用已经清理的记录不能用估算值冒充真实成本。

tests/test_admin_ad_revenue_scope.py

@@ -8,7 +8,6 @@ from sqlalchemy import delete
 from app.admin.repositories import ad_revenue
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
-from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_pangle_revenue import AdPangleDailyRevenue
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
@@ -209,112 +208,3 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         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"
-    detail_date = "2040-02-06"
-    try:
-        user = User(phone=phone, username="29999999994", register_channel="sms")
-        db.add(user)
-        db.flush()
-        db.add_all([
-            AdFeedRewardRecord(
-                client_event_id="detail-adn-pangle", user_id=user.id,
-                reward_date=detail_date, duration_seconds=20, unit_count=1,
-                ecpm_raw="12000", adn="pangle", slot_id="rit-pangle",
-                ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
-                app_env="prod", our_code_id="104098712", coin=12, status="granted",
-                created_at=datetime(2040, 2, 6, 1, tzinfo=UTC),
-            ),
-            AdFeedRewardRecord(
-                client_event_id="detail-adn-gdt", user_id=user.id,
-                reward_date=detail_date, duration_seconds=20, unit_count=1,
-                ecpm_raw="25000", adn="gdt", slot_id="rit-gdt",
-                ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
-                app_env="prod", our_code_id="104098712", coin=25, status="granted",
-                created_at=datetime(2040, 2, 6, 2, tzinfo=UTC),
-            ),
-        ])
-        db.commit()
-
-        result = ad_revenue.ad_revenue_report(
-            db, date_from=detail_date, date_to=detail_date,
-            user_id=user.id, app_env="prod", revenue_scope="all",
-        )
-
-        item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
-        assert [detail["adn"] for detail in item["sub_rewards"]] == ["pangle", "gdt"]
-        assert [detail["slot_id"] for detail in item["sub_rewards"]] == ["rit-pangle", "rit-gdt"]
-    finally:
-        db.rollback()
-        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == detail_date))
-        db.execute(delete(User).where(User.phone == phone))
-        db.commit()
-        db.close()
-
-
-def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
-    db = SessionLocal()
-    phone = "18800009995"
-    fallback_date = "2040-02-07"
-    try:
-        user = User(phone=phone, username="29999999995", register_channel="sms")
-        db.add(user)
-        db.flush()
-        db.add_all([
-            AdFeedRewardRecord(
-                client_event_id="source-fallback-unique", user_id=user.id,
-                reward_date=fallback_date, duration_seconds=3, unit_count=0,
-                ecpm_raw="4700", ad_session_id="flow-session", trace_id="source-trace",
-                app_env="prod", our_code_id="104098712", coin=0, status="too_short",
-                ad_type="draw", feed_scene="comparison",
-                created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
-            ),
-            AdFeedRewardRecord(
-                client_event_id="source-fallback-ambiguous", user_id=user.id,
-                reward_date=fallback_date, duration_seconds=3, unit_count=0,
-                ecpm_raw="4800", ad_session_id="flow-session", trace_id="source-trace",
-                app_env="prod", our_code_id="104098712", coin=0, status="too_short",
-                ad_type="draw", feed_scene="comparison",
-                created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
-            ),
-            AdEcpmRecord(
-                user_id=user.id, ad_type="draw", ad_session_id="impression-unique",
-                trace_id="source-trace", ecpm_raw="4700", adn="baidu", slot_id="rit-baidu",
-                app_env="prod", our_code_id="104098712", report_date=fallback_date,
-                created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
-            ),
-            AdEcpmRecord(
-                user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-a",
-                trace_id="source-trace", ecpm_raw="4800", adn="baidu", slot_id="rit-baidu",
-                app_env="prod", our_code_id="104098712", report_date=fallback_date,
-                created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
-            ),
-            AdEcpmRecord(
-                user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-b",
-                trace_id="source-trace", ecpm_raw="4800", adn="ks", slot_id="rit-ks",
-                app_env="prod", our_code_id="104098712", report_date=fallback_date,
-                created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
-            ),
-        ])
-        db.commit()
-
-        result = ad_revenue.ad_revenue_report(
-            db, date_from=fallback_date, date_to=fallback_date,
-            user_id=user.id, app_env="prod", revenue_scope="all",
-        )
-        item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
-        details = {detail["ecpm"]: detail for detail in item["sub_rewards"]}
-        assert details["4700"]["adn"] == "baidu"
-        assert details["4700"]["slot_id"] == "rit-baidu"
-        assert details["4800"]["adn"] is None
-        assert details["4800"]["slot_id"] is None
-    finally:
-        db.rollback()
-        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == fallback_date))
-        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == fallback_date))
-        db.execute(delete(User).where(User.phone == phone))
-        db.commit()
-        db.close()

tests/test_admin_read.py

@@ -1,7 +1,7 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import datetime
 
 import pytest
 from fastapi.testclient import TestClient
@@ -9,7 +9,6 @@ from sqlalchemy import event
 
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
-from app.admin.repositories import queries
 from app.db.session import SessionLocal, engine
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
@@ -145,8 +144,6 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     ) -> None:
         if "ad_reward_record.boost_round_id" in statement:
             raise AssertionError("提现详情不应查询未使用的 boost_round_id")
-        if "FROM user" in statement and "user.phone" in statement:
-            raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表")
 
     event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
     try:
@@ -165,17 +162,6 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     assert records.status_code == 200, records.text
 
 
-def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
-    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
-    naive = datetime(2038, 1, 1, 8, 0)
-    aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)
-
-    rows = [{"created_at": aware}, {"created_at": naive}]
-    rows.sort(key=queries._coin_record_sort_key, reverse=True)
-
-    assert rows == [{"created_at": naive}, {"created_at": aware}]
-
-
 def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
     _seed_user_with_data("13800000003")
     r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))

tests/test_compare_harvest.py

@@ -9,16 +9,17 @@ 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:
@@ -215,9 +216,7 @@ 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, patch(
-        "app.api.v1.compare.backfill_comparison_llm_cost"
-    ) as backfill:
+    with p:
         r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
     assert r.status_code == 200
     with SessionLocal() as db:
@@ -225,7 +224,6 @@ 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

@@ -1,161 +0,0 @@
-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_derive_best_excludes_short.py

@@ -1,102 +0,0 @@
-"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"。
-
-回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
-会把缺菜店的虚低价当 best_price → 记录页戴"最低"红框 + 虚假省额。
-修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)。
-纯函数, 不碰 DB。
-"""
-from app.repositories.comparison import _derive, _derive_from_results
-from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
-
-
-def test_derive_from_results_excludes_short_ordered_from_best():
-    # jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
-    results = [
-        {"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
-        {"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
-        {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
-    ]
-    platform_results = {
-        "jd_waimai": {"skipped_dish_count": 2},
-        "taobao_flash": {"skipped_dish_count": 0},
-        "meituan": {"is_source": True},
-    }
-    d = _derive_from_results(results, platform_results)
-    assert d["best_platform_id"] == "taobao_flash"
-    assert d["best_price_cents"] == 3850
-    assert d["saved_amount_cents"] == 4200 - 3850   # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
-    assert d["is_source_best"] is False
-
-
-def test_derive_from_results_all_targets_short_falls_back_to_source():
-    # 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
-    d = _derive_from_results(results, platform_results)
-    assert d["best_platform_id"] == "meituan"
-    assert d["is_source_best"] is True
-    assert d["saved_amount_cents"] == 0
-
-
-def test_derive_from_results_no_platform_results_keeps_old_behavior():
-    # 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    d = _derive_from_results(results)
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500
-
-
-def test_derive_from_results_malformed_platform_results_no_crash():
-    # 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
-    results = [
-        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
-        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
-    ]
-    d = _derive_from_results(results, {"jd_waimai": "oops"})
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500
-
-
-def test_derive_pydantic_excludes_short():
-    # _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
-    payload = ComparisonRecordIn(
-        trace_id="t-short-pyd",
-        source_price=42.0,
-        source_platform_id="meituan",
-        comparison_results=[
-            ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
-            ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
-            ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
-        ],
-        platform_results={
-            "jd_waimai": {"skipped_dish_count": 2},
-            "taobao_flash": {"skipped_dish_count": 0},
-        },
-    )
-    d = _derive(payload)
-    assert d["best_platform_id"] == "taobao_flash"
-    assert d["best_price_cents"] == 3850
-    assert d["saved_amount_cents"] == 4200 - 3850
-    assert d["is_source_best"] is False
-
-
-def test_derive_pydantic_malformed_platform_results_no_crash():
-    # _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
-    payload = ComparisonRecordIn(
-        trace_id="t-malformed-pyd",
-        source_price=42.0,
-        comparison_results=[
-            ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
-            ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
-        ],
-        platform_results={"jd_waimai": "oops"},
-    )
-    d = _derive(payload)  # 不抛 AttributeError
-    assert d["best_platform_id"] == "jd_waimai"
-    assert d["best_price_cents"] == 2500

tests/test_llm_cost.py

@@ -132,7 +132,6 @@ 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}},
@@ -140,12 +139,7 @@ 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(
-        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
-    )
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
+    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
 
     db = SessionLocal()
     try:

tests/test_pricebot_llm_calls.py

@@ -1,55 +0,0 @@
-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