功能：支持动态配置每日比价次数上限

修复：统一用户Draw信息流eCPM统计口径 (#190 )
## 问题业务收益详情的平均 Draw eCPM 仅平均成功发奖记录，会排除未发奖的真实展示，导致数值系统性偏高，且与广告收益页口径不一致。 ## 修复 - `feed_avg_ecpm` 改为从 `ad_ecpm_record` 的全部 `draw/feed` 实际展示计算 - 成功发奖、未发奖展示均纳入，每次展示等权 - 日期、正式/测试环境、业务代码位、领券/比价场景支持与广告收益页对齐 - 奖励份数仍基于成功发奖表，不混用展示数据源 - 复用广告收益报表的业务代码位集合 ## 线上数据复算 2026-07-25、正式业务、用户 #33： - 旧口径（只看成功发奖）：`29.9117 元/千次` - 新口径（333 次真实展示）：`19.9926 元/千次` - 新值与广告收益报表一致 ## 验证 - 新增成功/未发奖、场景、环境、业务代码位回归用例 - `tests/test_admin_read.py` + `tests/test_admin_ad_revenue_scope.py`：29 项全通过 - Ruff 改动文件检查通过 ## 上线顺序本 PR 需先于管理后台配套 PR 上线。 --------- Co-authored-by: guke <guke@wonderable.ai> Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #190 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-28 21:49:24 +08:00 · 2026-07-28 17:58:23 +08:00
11 changed files with 254 additions and 33 deletions
@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
 _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})


-def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
+def business_code_ids(db: Session, app_env: str | None) -> set[str]:
    """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
    prod_config = app_config.get_ad_config(db)
    prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(

    # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
    # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
-    business_code_ids: set[str] | None = None
+    business_ids: set[str] | None = None
    if revenue_scope == "business":
-        business_code_ids = _business_code_ids(db, app_env)
-        events = [e for e in events if e.get("our_code_id") in business_code_ids]
+        business_ids = business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_ids]

    # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
    # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
            date_from=date_from,
            date_to=date_to,
            app_env=app_env,
-            our_code_ids=business_code_ids,
+            our_code_ids=business_ids,
        )
        if pangle_aggs:
            by_date = {a["date"]: a for a in pangle_aggs}
@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session

 from app.core import rewards
 from app.core.config import settings
+from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
@@ -1257,11 +1258,15 @@ def user_reward_stats(
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    withdraw_source: str | None = None,
+    app_env: str | None = None,
+    revenue_scope: str = "all",
+    feed_scene: str | None = None,
 ) -> dict:
    """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。

-    口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
-    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
+    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
+    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
+    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
    传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
    """
    withdraw_source_conds = (
@@ -1296,30 +1301,68 @@ def user_reward_stats(

    # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
    # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
+    business_ids: set[str] | None = None
+    if revenue_scope == "business":
+        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
+        from app.admin.repositories.ad_revenue import business_code_ids
+
+        business_ids = business_code_ids(db, app_env)
+
+    rv_conds = [
+        AdRewardRecord.user_id == user_id,
+        AdRewardRecord.reward_scene == "reward_video",
+        AdRewardRecord.status == "granted",
+        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        rv_conds.append(AdRewardRecord.app_env == app_env)
+    if business_ids is not None:
+        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
    rv = db.execute(
        select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
-            AdRewardRecord.user_id == user_id,
-            AdRewardRecord.reward_scene == "reward_video",
-            AdRewardRecord.status == "granted",
-            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+            *rv_conds,
        )
    ).all()
    rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
    rv_coins = sum(r.coin for r in rv)

-    feed = db.execute(
+    feed_reward_conds = [
+        AdFeedRewardRecord.user_id == user_id,
+        AdFeedRewardRecord.status == "granted",
+        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
+    if feed_scene is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
+    if business_ids is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
+    feed_rewards = db.execute(
        select(
            AdFeedRewardRecord.unit_count,
-            AdFeedRewardRecord.ecpm_raw,
            AdFeedRewardRecord.coin,
        ).where(
-            AdFeedRewardRecord.user_id == user_id,
-            AdFeedRewardRecord.status == "granted",
-            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+            *feed_reward_conds,
        )
    ).all()
-    feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
-    feed_coins = sum(f.coin for f in feed)
+    feed_coins = sum(f.coin for f in feed_rewards)
+
+    feed_impression_conds = [
+        AdEcpmRecord.user_id == user_id,
+        AdEcpmRecord.ad_type.in_(("draw", "feed")),
+        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
+    if feed_scene is not None:
+        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
+    if business_ids is not None:
+        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
+    feed_impressions = db.execute(
+        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
+    ).all()
+    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
+    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]

    trad_coins = db.execute(
        select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1338,7 +1381,7 @@ def user_reward_stats(
        "reward_video_count": len(rv),
        "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
        "reward_video_cash_cents": _coins_to_cents(rv_coins),
-        "feed_count": int(sum(f.unit_count for f in feed)),
+        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
        "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
        "feed_cash_cents": _coins_to_cents(feed_coins),
    }
@@ -88,6 +88,11 @@ def get_user_reward_stats(
    withdraw_source: Annotated[
        str | None, Query(pattern="^(coin_cash|invite_cash)$")
    ] = None,
+    app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
+    revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
+    feed_scene: Annotated[
+        str | None, Query(pattern="^(comparison|coupon|welfare)$")
+    ] = None,
 ) -> UserRewardStats:
    """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
    if not user_repo.user_exists(db, user_id):
@@ -99,6 +104,9 @@ def get_user_reward_stats(
            date_from=date_from,
            date_to=date_to,
            withdraw_source=withdraw_source,
+            app_env=app_env,
+            revenue_scope=revenue_scope,
+            feed_scene=feed_scene,
        )
    )

@@ -23,7 +23,7 @@ class RiskMonitorSummary(BaseModel):
 class RiskRuleConfig(BaseModel):
    sms_hourly_threshold: int = Field(ge=1, le=5)
    oneclick_daily_threshold: int = Field(ge=1, le=100_000)
-    compare_daily_threshold: int = Field(ge=1, le=100)
+    compare_daily_threshold: int = Field(ge=1, le=100_000)


 class RiskIncidentItem(BaseModel):
@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
    reward_video_avg_ecpm: float  # 平均激励视频 eCPM(分/千次)
    reward_video_cash_cents: int  # 激励视频提现(金币折现)
    feed_count: int  # 累计信息流广告数(granted 份数,unit_count 累加)
-    feed_avg_ecpm: float  # 平均信息流广告 eCPM(分/千次)
+    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
    feed_cash_cents: int  # 信息流广告提现(金币折现)


@@ -38,13 +38,14 @@ router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
@router.post(
    "/start",
    response_model=CompareStartReserveOut,
-    summary="预占一次当日比价发起次数（每人每天最多100次）",
+    summary="预占一次当日比价发起次数（上限由风控监控配置）",
 )
 def reserve_compare_start(
    payload: CompareStartReserveIn,
    user: CurrentUser,
    db: DbSession,
 ) -> CompareStartReserveOut:
+    daily_limit = risk_repo.get_rule_threshold(db, risk_repo.RULE_COMPARE_DAILY)
    if risk_repo.is_restricted(
        db,
        subject_type="user",
@@ -57,25 +58,26 @@ def reserve_compare_start(
            db,
            user_id=user.id,
            trace_id=payload.trace_id,
+            daily_limit=daily_limit,
            business_type=payload.business_type,
            device_id=payload.device_id,
        )
    except crud_compare.DailyCompareStartLimitExceeded:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail="今日已比价超过100次，请明天再试",
+            detail=f"今日已比价超过{daily_limit}次，请明天再试",
        ) from None
    except crud_compare.ComparisonTraceOwnershipError:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="比价任务标识冲突，请重新发起",
        ) from None
-    # 风控阈值由后台动态配置，不能再只在固定的 100 次业务上限处同步。
+    # 告警和业务限流共用同一动态阈值，避免后台已修改但用户侧仍固定为 100 次。
    risk_repo.sync_compare_incident(db, user_id=user.id, at=rec.created_at)
    return CompareStartReserveOut(
-        limit=crud_compare.DAILY_COMPARE_START_LIMIT,
+        limit=daily_limit,
        used=used,
-        remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
+        remaining=max(daily_limit - used, 0),
    )


@@ -134,12 +134,12 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
    },
    RISK_COMPARE_DAILY_THRESHOLD_KEY: {
        "default": 100,
-        "label": "比价账户每日告警阈值",
+        "label": "比价账户每日上限与告警阈值",
        "group": "风控",
        "type": "int",
        "min": 1,
-        "max": 100,
+        "max": 100_000,
        "hidden": True,
-        "help": "同一账户在北京时间同一自然日内发起比价达到该次数时告警。",
+        "help": "同一账户按北京时间自然日最多可发起的比价次数；达到该次数时同步告警。",
    },
 }
@@ -17,8 +17,6 @@ from app.models.savings import SavingsRecord
 from app.models.user import User
 from app.schemas.compare_record import ComparisonRecordIn

-DAILY_COMPARE_START_LIMIT = 100
-

 class DailyCompareStartLimitExceeded(Exception):
    """The authenticated user has consumed today's comparison-start quota."""
@@ -363,11 +361,12 @@ def reserve_daily_start(
    *,
    user_id: int,
    trace_id: str,
+    daily_limit: int,
    business_type: str = "food",
    device_id: str | None = None,
    now: datetime | None = None,
 ) -> tuple[ComparisonRecord, int]:
-    """Atomically reserve one of a user's 100 Beijing-day comparison starts.
+    """Atomically reserve one of a user's configured Beijing-day comparison starts.

    ``trace_id`` makes client retries idempotent. Locking the user row serializes
    concurrent starts for one account, so parallel requests cannot both consume
@@ -413,7 +412,7 @@ def reserve_daily_start(
            ComparisonRecord.created_at < day_end,
        )
    ) or 0
-    if used >= DAILY_COMPARE_START_LIMIT:
+    if used >= daily_limit:
        raise DailyCompareStartLimitExceeded

    rec = ComparisonRecord(
@@ -219,6 +219,115 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
    assert invite.json()["cash_balance_cents"] == 456


+def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
+    from app.models.ad_ecpm import AdEcpmRecord
+    from app.models.ad_feed_reward import AdFeedRewardRecord
+
+    uid = _seed_user_with_data("13800000024")
+    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
+    db = SessionLocal()
+    try:
+        db.add_all(
+            [
+                AdFeedRewardRecord(
+                    client_event_id="reward-stats-granted-high",
+                    ad_session_id="reward-stats-granted-high",
+                    user_id=uid,
+                    reward_date="2038-01-15",
+                    duration_seconds=10,
+                    unit_count=1,
+                    ecpm_raw="9000",
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    coin=9,
+                    status="granted",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-impression-low",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="1000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="feed",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-impression-mid",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="3000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="comparison",
+                    ad_session_id="reward-stats-other-scene",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="7000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-test-env",
+                    app_env="test",
+                    our_code_id="104127529",
+                    ecpm_raw="8000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-non-business",
+                    app_env="prod",
+                    our_code_id="demo-slot",
+                    ecpm_raw="9000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+            ]
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        f"/admin/api/users/{uid}/reward-stats",
+        params={
+            "date_from": "2038-01-15T00:00:00Z",
+            "date_to": "2038-01-15T23:59:59Z",
+            "app_env": "prod",
+            "revenue_scope": "business",
+            "feed_scene": "coupon",
+        },
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert data["feed_count"] == 1
+    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
+    assert data["feed_avg_ecpm"] == 2000.0
+
+
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
    naive = datetime(2038, 1, 1, 8, 0)
@@ -3,14 +3,32 @@ from __future__ import annotations
 import time
 from datetime import datetime, timedelta

+import pytest
 from sqlalchemy import func, select

+from app.core.config_schema import RISK_COMPARE_DAILY_THRESHOLD_KEY
 from app.core.rewards import CN_TZ
 from app.core.security import decode_token
 from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
 from app.models.comparison import ComparisonRecord


+@pytest.fixture(autouse=True)
+def _reset_compare_daily_limit():
+    with SessionLocal() as db:
+        db.query(AppConfig).filter(
+            AppConfig.key == RISK_COMPARE_DAILY_THRESHOLD_KEY
+        ).delete()
+        db.commit()
+    yield
+    with SessionLocal() as db:
+        db.query(AppConfig).filter(
+            AppConfig.key == RISK_COMPARE_DAILY_THRESHOLD_KEY
+        ).delete()
+        db.commit()
+
+
 def _login(client) -> tuple[str, int]:
    phone = f"137{int(time.time() * 1000) % 100000000:08d}"
    sent = client.post("/api/v1/auth/sms/send", json={"phone": phone})
@@ -118,3 +136,33 @@ def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
                ComparisonRecord.trace_id == rejected_trace
            )
        ) == 0
+
+
+def test_compare_start_uses_dynamic_risk_monitor_limit(client) -> None:
+    with SessionLocal() as db:
+        db.add(AppConfig(key=RISK_COMPARE_DAILY_THRESHOLD_KEY, value=2))
+        db.commit()
+
+    token, user_id = _login(client)
+    first = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": f"quota-dynamic-{user_id}-1"},
+        headers=_headers(token),
+    )
+    second = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": f"quota-dynamic-{user_id}-2"},
+        headers=_headers(token),
+    )
+    rejected = client.post(
+        "/api/v1/compare/start",
+        json={"trace_id": f"quota-dynamic-{user_id}-3"},
+        headers=_headers(token),
+    )
+
+    assert first.status_code == 200
+    assert first.json() == {"limit": 2, "used": 1, "remaining": 1}
+    assert second.status_code == 200
+    assert second.json() == {"limit": 2, "used": 2, "remaining": 0}
+    assert rejected.status_code == 429
+    assert rejected.json()["detail"] == "今日已比价超过2次，请明天再试"
@@ -529,6 +529,18 @@ def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
                now + timedelta(seconds=2)
            ).replace(tzinfo=None)

+        compare_limit = client.patch(
+            "/admin/api/risk-monitor/rules",
+            headers=headers,
+            json={
+                "sms_hourly_threshold": 3,
+                "oneclick_daily_threshold": 20,
+                "compare_daily_threshold": 120,
+            },
+        )
+        assert compare_limit.status_code == 200
+        assert compare_limit.json()["compare_daily_threshold"] == 120
+
        invalid = client.patch(
            "/admin/api/risk-monitor/rules",
            headers=headers,