fix: admin 比价记录列表兼容 user_id 为空的孤儿行

软鉴权/匿名下 pricebot 帧0 建行时 user_id 可空(见 models.comparison),admin
全看含孤儿行;但 AdminComparisonListItem.user_id 声明为必填 int,Pydantic v2
对 None 抛 ValidationError,致 GET /admin/api/comparison-records 列表接口 500。
改为 int|None(详情接口继承一并修复),并补回归测试
test_comparison_records_list_tolerates_orphan_null_user。

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

alembic/versions/deepseek_v4_flash_price.py

@@ -0,0 +1,211 @@
+"""correct DeepSeek V4 Flash token price and frozen historical costs
+
+Revision ID: deepseek_v4_flash_price
+Revises: comparison_below_min_success
+"""
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from typing import Any
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "deepseek_v4_flash_price"
+down_revision: str | Sequence[str] | None = "comparison_below_min_success"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+_CONFIG_KEY = "llm_token_price"
+_MODEL = "deepseek-v4-flash"
+_OLD_INPUT_PRICE = 3.0
+_OLD_OUTPUT_PRICE = 15.0
+_NEW_INPUT_PRICE = 1.0
+_NEW_OUTPUT_PRICE = 2.0
+_CORRECTION_MARKER = "deepseek_v4_flash_price"
+_CONFIG_MARKER_KEY = "migration_deepseek_v4_flash_price"
+
+
+def _decode_object(value: Any) -> dict[str, Any] | None:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except (TypeError, ValueError):
+            return None
+        return parsed if isinstance(parsed, dict) else None
+    return None
+
+
+def _model_tokens(calls: Any) -> tuple[int, int]:
+    if not isinstance(calls, list):
+        return 0, 0
+    input_tokens = 0
+    output_tokens = 0
+    for call in calls:
+        if not isinstance(call, dict) or call.get("error") or call.get("model") != _MODEL:
+            continue
+        usage = _decode_object(call.get("usage"))
+        if usage is None:
+            continue
+        input_tokens += int(usage.get("prompt_tokens") or 0)
+        output_tokens += int(usage.get("completion_tokens") or 0)
+    return input_tokens, output_tokens
+
+
+def _app_config_table() -> sa.TableClause:
+    return sa.table(
+        "app_config",
+        sa.column("key", sa.String(64)),
+        sa.column("value", _JSON),
+        sa.column("updated_at", sa.DateTime(timezone=True)),
+    )
+
+
+def _comparison_table() -> sa.TableClause:
+    return sa.table(
+        "comparison_record",
+        sa.column("id", sa.Integer),
+        sa.column("llm_calls", _JSON),
+        sa.column("llm_cost_yuan", sa.Float),
+        sa.column("llm_price_snapshot", _JSON),
+    )
+
+
+def _update_config(conn, *, upgrade: bool) -> None:
+    table = _app_config_table()
+    row = conn.execute(
+        sa.select(table.c.value).where(table.c.key == _CONFIG_KEY)
+    ).mappings().first()
+    if row is None or not isinstance(row["value"], dict):
+        return
+
+    config = dict(row["value"])
+    per_model = dict(config.get("per_model") or {})
+    current = per_model.get(_MODEL)
+    if upgrade:
+        # Preserve an operator's explicit model price. The production defect is specifically
+        # the missing key falling through to the generic 3/15 price.
+        if current is not None:
+            return
+        per_model[_MODEL] = {
+            "input_per_1m": _NEW_INPUT_PRICE,
+            "output_per_1m": _NEW_OUTPUT_PRICE,
+        }
+        conn.execute(table.insert().values(key=_CONFIG_MARKER_KEY, value=True))
+    else:
+        marker_exists = conn.execute(
+            sa.select(table.c.key).where(table.c.key == _CONFIG_MARKER_KEY)
+        ).scalar_one_or_none()
+        if marker_exists is None:
+            return
+        if current != {
+            "input_per_1m": _NEW_INPUT_PRICE,
+            "output_per_1m": _NEW_OUTPUT_PRICE,
+        }:
+            conn.execute(table.delete().where(table.c.key == _CONFIG_MARKER_KEY))
+            return
+        per_model.pop(_MODEL, None)
+        conn.execute(table.delete().where(table.c.key == _CONFIG_MARKER_KEY))
+    config["per_model"] = per_model
+    conn.execute(
+        table.update()
+        .where(table.c.key == _CONFIG_KEY)
+        # 这是对历史误配置的追溯修正，不是从部署时刻开始的新价格。保留原 updated_at，
+        # 否则缺失成本回填会把部署前的记录全部排除。
+        .values(value=config)
+    )
+
+
+def _correct_frozen_costs(conn, *, upgrade: bool) -> None:
+    table = _comparison_table()
+    rows = conn.execute(
+        sa.select(
+            table.c.id,
+            table.c.llm_calls,
+            table.c.llm_cost_yuan,
+            table.c.llm_price_snapshot,
+        ).where(
+            table.c.llm_cost_yuan.is_not(None),
+            table.c.llm_price_snapshot.is_not(None),
+        )
+    ).mappings()
+
+    for row in rows:
+        snapshot = row["llm_price_snapshot"]
+        if not isinstance(snapshot, dict):
+            continue
+        prices = snapshot.get("prices")
+        if not isinstance(prices, dict):
+            continue
+        model_price = prices.get(_MODEL)
+        if not isinstance(model_price, dict):
+            continue
+
+        if upgrade:
+            if not (
+                model_price.get("_source") == "default"
+                and model_price.get("input_per_1m") == _OLD_INPUT_PRICE
+                and model_price.get("output_per_1m") == _OLD_OUTPUT_PRICE
+            ):
+                continue
+        elif snapshot.get("pricing_correction") != _CORRECTION_MARKER:
+            continue
+
+        input_tokens, output_tokens = _model_tokens(row["llm_calls"])
+        if input_tokens == 0 and output_tokens == 0:
+            continue
+        if upgrade:
+            delta = (
+                input_tokens / 1_000_000 * (_OLD_INPUT_PRICE - _NEW_INPUT_PRICE)
+                + output_tokens / 1_000_000 * (_OLD_OUTPUT_PRICE - _NEW_OUTPUT_PRICE)
+            )
+            corrected_price = {
+                "input_per_1m": _NEW_INPUT_PRICE,
+                "output_per_1m": _NEW_OUTPUT_PRICE,
+                "_source": "per_model",
+            }
+            snapshot["pricing_correction"] = _CORRECTION_MARKER
+            new_cost = max(0.0, float(row["llm_cost_yuan"]) - delta)
+        else:
+            delta = (
+                input_tokens / 1_000_000 * (_OLD_INPUT_PRICE - _NEW_INPUT_PRICE)
+                + output_tokens / 1_000_000 * (_OLD_OUTPUT_PRICE - _NEW_OUTPUT_PRICE)
+            )
+            corrected_price = {
+                "input_per_1m": _OLD_INPUT_PRICE,
+                "output_per_1m": _OLD_OUTPUT_PRICE,
+                "_source": "default",
+            }
+            snapshot.pop("pricing_correction", None)
+            new_cost = float(row["llm_cost_yuan"]) + delta
+
+        updated_prices = dict(prices)
+        updated_prices[_MODEL] = corrected_price
+        updated_snapshot = dict(snapshot)
+        updated_snapshot["prices"] = updated_prices
+        conn.execute(
+            table.update()
+            .where(table.c.id == row["id"])
+            .values(
+                llm_cost_yuan=round(new_cost, 6),
+                llm_price_snapshot=updated_snapshot,
+            )
+        )
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    _update_config(conn, upgrade=True)
+    _correct_frozen_costs(conn, upgrade=True)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    _correct_frozen_costs(conn, upgrade=False)
+    _update_config(conn, upgrade=False)

app/admin/repositories/coupon_data.py

@@ -166,6 +166,18 @@ def _session_to_row(
     point_stats: dict | None = None,
 ) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
+    # 中途退出可能发生在第一张券产生终态之前，此时没有逐券事件。
+    # 明确返回 0/0，让前端区分「退出前无单券结果」与其它状态的埋点缺失。
+    if point_stats is not None:
+        point_success_count = point_stats["succeeded"]
+        point_total_count = point_stats["tried"]
+    elif r.status == "abandoned":
+        point_success_count = 0
+        point_total_count = 0
+    else:
+        point_success_count = None
+        point_total_count = None
+    point_event_count = point_stats["events"] if point_stats is not None else 0
     return {
         "id": r.id,
         "trace_id": r.trace_id,
@@ -182,8 +194,9 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": point_stats["succeeded"] if point_stats else None,
+        "point_success_count": point_success_count,
-        "point_total_count": point_stats["tried"] if point_stats else None,
+        "point_total_count": point_total_count,
+        "point_event_count": point_event_count,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
@@ -194,21 +207,24 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
     if not trace_ids:
         return {}
     succeeded = func.sum(case((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
+    tried = func.sum(case((CouponClaimEvent.status.in_(_SLOT_TRIED), 1), else_=0))
     rows = db.execute(
         select(
             CouponClaimEvent.trace_id,
             succeeded.label("succeeded"),
-            func.count().label("tried"),
+            tried.label("tried"),
-        )
+            func.count().label("events"),
-        .where(
-            CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimEvent.status.in_(_SLOT_TRIED),
         )
+        .where(CouponClaimEvent.trace_id.in_(trace_ids))
         .group_by(CouponClaimEvent.trace_id)
     ).all()
     return {
-        trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
+        trace_id: {
-        for trace_id, success_count, tried in rows
+            "succeeded": int(success_count or 0),
+            "tried": int(tried_count or 0),
+            "events": int(event_count or 0),
+        }
+        for trace_id, success_count, tried_count, event_count in rows
         if trace_id is not None
     }
 
@@ -400,12 +416,15 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
     total = db.execute(
         select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
     ).scalar_one()
-    rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
+    trace_ids = [r.trace_id for r in rows]
+    rev_map = crud_ecpm.revenue_yuan_by_trace(db, trace_ids)
+    point_stats_map = _point_scores_by_trace(db, trace_ids)
     return {
         "items": [
             _session_to_row(
                 r,
                 ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
+                point_stats=point_stats_map.get(r.trace_id),
             )
             for r in rows
         ],

app/admin/repositories/stats.py

@@ -566,6 +566,10 @@ def dashboard_overview(
         )
     ).all()
     coupon_started = len(period_coupon_sessions)
+    coupon_abandoned = sum(s.status == "abandoned" for s in period_coupon_sessions)
+    # 用户主动中途退出不代表领券流程失败，不进入整场成功率样本。
+    # started / failed 仍留在分母：前者是尚未形成终态的流失，后者是实际执行失败。
+    coupon_success_denominator = coupon_started - coupon_abandoned
     coupon_completed_elapsed = sorted(
         s.elapsed_ms
         for s in period_coupon_sessions
@@ -731,9 +735,13 @@ def dashboard_overview(
             },
             "coupon": {
                 "started": coupon_started,
+                "abandoned": coupon_abandoned,
+                "success_denominator": coupon_success_denominator,
                 "all_success": coupon_all_success,
                 "success_rate": (
-                    round(coupon_all_success / coupon_started, 4) if coupon_started else None
+                    round(coupon_all_success / coupon_success_denominator, 4)
+                    if coupon_success_denominator
+                    else None
                 ),
                 "point_success": coupon_point_success,
                 "points_per_session": coupon_points_per_session,

app/admin/schemas/comparison.py

@@ -13,7 +13,8 @@ class AdminComparisonListItem(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
-    user_id: int
+    # 软鉴权/匿名下帧0 建行时 user_id 可能暂缺(见 models.comparison 注释);admin 全看含孤儿行,故可空。
+    user_id: int | None = None
     phone: str | None = None        # join User 瞬态(非 DB 列)
     nickname: str | None = None     # join User 瞬态
     business_type: str

app/admin/schemas/coupon_data.py

@@ -79,10 +79,16 @@ class CouponDataRow(BaseModel):
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
     point_success_count: int | None = Field(
-        None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
+        None,
+        description="本次成功单券数(success+already_claimed);中途退出且无逐券结果为0,其它无事件为空",
     )
     point_total_count: int | None = Field(
-        None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
+        None,
+        description="本次尝试单券数(success+already_claimed+failed,不含 skipped);中途退出且无逐券结果为0,其它无事件为空",
+    )
+    point_event_count: int = Field(
+        0,
+        description="本次全部逐券事件数(含 skipped);用于区分无有效计分事件与完全无事件",
     )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(

app/admin/schemas/dashboard.py

@@ -70,6 +70,9 @@ class DashboardPeriodCoupon(BaseModel):
     成功口径 success+already_claimed(与「我的」页累计领券一致)。"""
 
     started: int = 0
+    # 用户主动中途退出，不计入整场成功率分母。
+    abandoned: int = 0
+    success_denominator: int = 0
     # 全部领成功的次数:completed 且当日该设备全部点位成功
     all_success: int = 0
     success_rate: float | None = None

app/core/config_schema.py

@@ -227,7 +227,11 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     # 编辑框;set_value 不校验类型,嵌套 JSON 照存。
     "llm_token_price": {
         "default": {
-            "per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
+            "per_model": {
+                "qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0},
+                # DashScope 华北 2 公网调用原价；必须显式配置，不能落到 3/15 的未知模型兜底价。
+                "deepseek-v4-flash": {"input_per_1m": 1.0, "output_per_1m": 2.0},
+            },
             "default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
             "currency": "CNY", "unit": "per_1m_tokens",
         },

tests/test_admin_read.py

@@ -1,7 +1,7 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime
 
 import pytest
 from fastapi.testclient import TestClient
@@ -12,6 +12,7 @@ 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.coupon_state import CouponClaimRecord, CouponSession
 from app.models.feedback import Feedback
 from app.models.invite import InviteRelation
 from app.models.savings import SavingsRecord
@@ -120,6 +121,86 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
     assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
 
 
+def test_dashboard_coupon_success_rate_excludes_abandoned_sessions(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    started_date = date(2038, 1, 16)
+    started_at = datetime(2038, 1, 16, 8, tzinfo=UTC)
+    sessions = [
+        ("coupon-rate-completed-1", "coupon-rate-device-1", "completed"),
+        ("coupon-rate-completed-2", "coupon-rate-device-2", "completed"),
+        ("coupon-rate-failed", "coupon-rate-device-3", "failed"),
+        ("coupon-rate-abandoned", "coupon-rate-device-4", "abandoned"),
+    ]
+    db = SessionLocal()
+    try:
+        for trace_id, device_id, status in sessions:
+            db.add(
+                CouponSession(
+                    trace_id=trace_id,
+                    device_id=device_id,
+                    status=status,
+                    app_env="prod",
+                    platforms=["meituan-waimai"],
+                    started_at=started_at,
+                    started_date=started_date,
+                )
+            )
+        for index, device_id in enumerate(("coupon-rate-device-1", "coupon-rate-device-2")):
+            db.add(
+                CouponClaimRecord(
+                    device_id=device_id,
+                    coupon_id=f"mt_dashboard_rate_{index}",
+                    claim_date=started_date,
+                    status="success",
+                    app_env="prod",
+                )
+            )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2038-01-16", "date_to": "2038-01-16"},
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    coupon = response.json()["period"]["coupon"]
+    assert coupon["started"] == 4
+    assert coupon["abandoned"] == 1
+    assert coupon["success_denominator"] == 3
+    assert coupon["all_success"] == 2
+    assert coupon["success_rate"] == pytest.approx(2 / 3, abs=0.0001)
+
+    db = SessionLocal()
+    try:
+        db.add(
+            CouponSession(
+                trace_id="coupon-rate-only-abandoned",
+                device_id="coupon-rate-device-only-abandoned",
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2038, 1, 17, 8, tzinfo=UTC),
+                started_date=date(2038, 1, 17),
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    empty_denominator_response = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2038-01-17", "date_to": "2038-01-17"},
+        headers=_auth(admin_token),
+    )
+    assert empty_denominator_response.status_code == 200
+    only_abandoned = empty_denominator_response.json()["period"]["coupon"]
+    assert only_abandoned["success_denominator"] == 0
+    assert only_abandoned["success_rate"] is None
+
+
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
     uid = _seed_user_with_data("13800000002")
     r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
@@ -761,6 +842,36 @@ def test_comparison_records_show_readable_device_and_rom_version(
     assert detail.json()["platforms"][0]["is_best"] is True
 
 
+def test_comparison_records_list_tolerates_orphan_null_user(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """帧0 建行但 user_id 暂缺的孤儿记录(软鉴权/匿名),admin 列表必须能序列化、不 500。"""
+    db = SessionLocal()
+    try:
+        db.add(
+            ComparisonRecord(
+                user_id=None,
+                trace_id="comparison-orphan-null-user",
+                status="success",
+                store_name="孤儿比价专用店ZZZ",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/comparison-records",
+        params={"store": "孤儿比价专用店ZZZ"},
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    items = response.json()["items"]
+    assert len(items) == 1
+    assert items[0]["user_id"] is None
+    assert items[0]["trace_id"] == "comparison-orphan-null-user"
+
+
 def test_comparison_records_show_real_order_status(
     admin_client: TestClient, admin_token: str
 ) -> None:

tests/test_coupon_point_score.py

@@ -10,6 +10,7 @@ from app.admin.repositories.coupon_data import (
     _point_scores_by_trace,
     coupon_data_report,
     coupon_point_details,
+    coupon_user_records,
 )
 from app.admin.security import create_admin_token
 from app.db.session import SessionLocal
@@ -38,6 +39,7 @@ def test_point_scores_by_trace() -> None:
         stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
         assert stats["tried"] == 3
+        assert stats["events"] == 4
         details = coupon_point_details(db, trace_id=trace)
         assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
@@ -48,8 +50,8 @@ def test_point_scores_by_trace() -> None:
         db.close()
 
 
-def test_skipped_detail_does_not_create_a_score() -> None:
+def test_skipped_detail_is_distinguished_from_no_events() -> None:
-    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
+    """仅有 skipped 时分数仍为0/0，但保留事件数供前端开放明细。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -63,7 +65,7 @@ def test_skipped_detail_does_not_create_a_score() -> None:
         db.flush()
 
         scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert trace not in scores
+        assert scores[trace] == {"succeeded": 0, "tried": 0, "events": 1}
         assert "missing-trace" not in scores
         assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
     finally:
@@ -114,6 +116,112 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
         db.close()
 
 
+def test_coupon_data_report_marks_abandoned_without_point_results() -> None:
+    """中途退出且没有逐券终态时返回0/0，其他状态缺埋点仍保持为空。"""
+    db = SessionLocal()
+    report_date = date(2020, 1, 6)
+    user_id = 910006
+    try:
+        db.add_all([
+            CouponSession(
+                trace_id="point-score-abandoned-without-result",
+                device_id="score-abandoned-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id="point-score-completed-without-result",
+                device_id="score-completed-device",
+                user_id=user_id,
+                status="completed",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, 1, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id="point-score-abandoned-with-result",
+                device_id="score-abandoned-result-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, 2, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id="point-score-abandoned-skipped-only",
+                device_id="score-abandoned-skipped-device",
+                user_id=user_id,
+                status="abandoned",
+                app_env="prod",
+                platforms=["meituan-waimai"],
+                started_at=datetime(2020, 1, 6, 3, tzinfo=UTC),
+                started_date=report_date,
+            ),
+        ])
+        db.add_all([
+            CouponClaimEvent(
+                trace_id="point-score-abandoned-with-result",
+                device_id="score-abandoned-result-device",
+                coupon_id=f"mt-abandoned-{status}",
+                claim_date=report_date,
+                status=status,
+            )
+            for status in ("success", "failed")
+        ])
+        db.add(CouponClaimEvent(
+            trace_id="point-score-abandoned-skipped-only",
+            device_id="score-abandoned-skipped-device",
+            coupon_id="mt-abandoned-skipped",
+            claim_date=report_date,
+            status="skipped",
+        ))
+        db.flush()
+
+        report = coupon_data_report(
+            db,
+            date_from=report_date.isoformat(),
+            date_to=report_date.isoformat(),
+            app_env="prod",
+        )
+        rows = {item["trace_id"]: item for item in report["items"]}
+        abandoned = rows["point-score-abandoned-without-result"]
+        assert abandoned["point_success_count"] == 0
+        assert abandoned["point_total_count"] == 0
+        assert abandoned["point_event_count"] == 0
+
+        abandoned_with_result = rows["point-score-abandoned-with-result"]
+        assert abandoned_with_result["point_success_count"] == 1
+        assert abandoned_with_result["point_total_count"] == 2
+        assert abandoned_with_result["point_event_count"] == 2
+
+        abandoned_skipped = rows["point-score-abandoned-skipped-only"]
+        assert abandoned_skipped["point_success_count"] == 0
+        assert abandoned_skipped["point_total_count"] == 0
+        assert abandoned_skipped["point_event_count"] == 1
+
+        completed = rows["point-score-completed-without-result"]
+        assert completed["point_success_count"] is None
+        assert completed["point_total_count"] is None
+
+        user_rows = {
+            item["trace_id"]: item
+            for item in coupon_user_records(db, user_id=user_id)["items"]
+        }
+        assert user_rows["point-score-abandoned-without-result"]["point_total_count"] == 0
+        assert user_rows["point-score-abandoned-with-result"]["point_total_count"] == 2
+        assert user_rows["point-score-abandoned-skipped-only"]["point_event_count"] == 1
+        assert user_rows["point-score-completed-without-result"]["point_total_count"] is None
+    finally:
+        db.rollback()
+        db.close()
+
+
 def test_coupon_point_details_endpoint() -> None:
     """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
     db = SessionLocal()

tests/test_deepseek_price_migration.py

@@ -0,0 +1,184 @@
+from __future__ import annotations
+
+import importlib.util
+from datetime import datetime
+from pathlib import Path
+
+import sqlalchemy as sa
+
+
+def _load_migration():
+    path = (
+        Path(__file__).parents[1]
+        / "alembic"
+        / "versions"
+        / "deepseek_v4_flash_price.py"
+    )
+    spec = importlib.util.spec_from_file_location("deepseek_v4_flash_price", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_migration_follows_comparison_status_normalization():
+    migration = _load_migration()
+
+    assert migration.down_revision == "comparison_below_min_success"
+
+
+def test_migration_adds_price_and_corrects_only_mispriced_snapshot(monkeypatch):
+    migration = _load_migration()
+    engine = sa.create_engine("sqlite://")
+    metadata = sa.MetaData()
+    app_config = sa.Table(
+        "app_config",
+        metadata,
+        sa.Column("key", sa.String(64), primary_key=True),
+        sa.Column("value", sa.JSON, nullable=False),
+        sa.Column("updated_at", sa.DateTime),
+    )
+    comparison = sa.Table(
+        "comparison_record",
+        metadata,
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("llm_calls", sa.JSON),
+        sa.Column("llm_cost_yuan", sa.Float),
+        sa.Column("llm_price_snapshot", sa.JSON),
+    )
+    metadata.create_all(engine)
+
+    calls = [
+        {
+            "model": "deepseek-v4-flash",
+            "error": None,
+            "usage": {"prompt_tokens": 12031, "completion_tokens": 125},
+        },
+        {
+            "model": "qwen3.5-flash",
+            "error": None,
+            "usage": {"prompt_tokens": 2416, "completion_tokens": 119},
+        },
+    ]
+    snapshot = {
+        "mode": "per_model",
+        "prices": {
+            "deepseek-v4-flash": {
+                "input_per_1m": 3.0,
+                "output_per_1m": 15.0,
+                "_source": "default",
+            },
+            "qwen3.5-flash": {
+                "input_per_1m": 0.2,
+                "output_per_1m": 2.0,
+                "_source": "per_model",
+            },
+        },
+    }
+
+    with engine.begin() as conn:
+        original_updated_at = datetime(2026, 7, 13, 18, 20, 10)
+        conn.execute(
+            app_config.insert().values(
+                key="llm_token_price",
+                value={
+                    "per_model": {
+                        "qwen3.5-flash": {
+                            "input_per_1m": 0.2,
+                            "output_per_1m": 2.0,
+                        }
+                    },
+                    "default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
+                },
+                updated_at=original_updated_at,
+            )
+        )
+        conn.execute(
+            comparison.insert().values(
+                id=1,
+                llm_calls=calls,
+                llm_cost_yuan=0.038689,
+                llm_price_snapshot=snapshot,
+            )
+        )
+        monkeypatch.setattr(migration.op, "get_bind", lambda: conn)
+
+        migration.upgrade()
+
+        config = conn.execute(
+            sa.select(app_config.c.value).where(
+                app_config.c.key == "llm_token_price"
+            )
+        ).scalar_one()
+        assert config["per_model"]["deepseek-v4-flash"] == {
+            "input_per_1m": 1.0,
+            "output_per_1m": 2.0,
+        }
+        assert conn.execute(
+            sa.select(app_config.c.updated_at).where(
+                app_config.c.key == "llm_token_price"
+            )
+        ).scalar_one() == original_updated_at
+        corrected = conn.execute(sa.select(comparison)).mappings().one()
+        assert corrected["llm_cost_yuan"] == 0.013002
+        assert corrected["llm_price_snapshot"]["prices"]["deepseek-v4-flash"] == {
+            "input_per_1m": 1.0,
+            "output_per_1m": 2.0,
+            "_source": "per_model",
+        }
+        assert corrected["llm_price_snapshot"]["pricing_correction"] == (
+            "deepseek_v4_flash_price"
+        )
+
+        migration.downgrade()
+
+        reverted_config = conn.execute(
+            sa.select(app_config.c.value).where(
+                app_config.c.key == "llm_token_price"
+            )
+        ).scalar_one()
+        assert "deepseek-v4-flash" not in reverted_config["per_model"]
+        reverted = conn.execute(sa.select(comparison)).mappings().one()
+        assert reverted["llm_cost_yuan"] == 0.038689
+        assert "pricing_correction" not in reverted["llm_price_snapshot"]
+
+
+def test_downgrade_preserves_price_that_existed_before_upgrade(monkeypatch):
+    migration = _load_migration()
+    engine = sa.create_engine("sqlite://")
+    metadata = sa.MetaData()
+    app_config = sa.Table(
+        "app_config",
+        metadata,
+        sa.Column("key", sa.String(64), primary_key=True),
+        sa.Column("value", sa.JSON, nullable=False),
+        sa.Column("updated_at", sa.DateTime),
+    )
+    sa.Table(
+        "comparison_record",
+        metadata,
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column("llm_calls", sa.JSON),
+        sa.Column("llm_cost_yuan", sa.Float),
+        sa.Column("llm_price_snapshot", sa.JSON),
+    )
+    metadata.create_all(engine)
+
+    explicit_price = {"input_per_1m": 1.0, "output_per_1m": 2.0}
+    with engine.begin() as conn:
+        conn.execute(
+            app_config.insert().values(
+                key="llm_token_price",
+                value={
+                    "per_model": {"deepseek-v4-flash": explicit_price},
+                    "default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
+                },
+            )
+        )
+        monkeypatch.setattr(migration.op, "get_bind", lambda: conn)
+
+        migration.upgrade()
+        migration.downgrade()
+
+        config = conn.execute(sa.select(app_config.c.value)).scalar_one()
+        assert config["per_model"]["deepseek-v4-flash"] == explicit_price

tests/test_llm_cost.py

@@ -108,6 +108,10 @@ def test_get_llm_prices_falls_back_to_default_then_uses_override():
         # 无 override → CONFIG_DEFS 默认(含 per_model / default)
         prices = get_llm_prices(db)
         assert "per_model" in prices and "default" in prices
+        assert prices["per_model"]["deepseek-v4-flash"] == {
+            "input_per_1m": 1.0,
+            "output_per_1m": 2.0,
+        }
         # 有 override → 用 DB 值
         app_config.set_value(
             db, "llm_token_price",