从零实现优惠券发放算法

主题

第 10 章 防薅羊毛风控 ​

对应源码: src/risk/anti_fraud.py

优惠券的最大敌人不是预算不够，而是"薅羊毛"。一个设备注册 10 个账号、一个人一天领 20 张券——如果不做风控，再好的 Uplift 模型也挡不住预算被薅光。

10.1 薅羊毛的典型手法 ​

在电商新用户优惠券场景下，常见的薅羊毛方式有：

手法描述危害
一机多号同一设备注册多个新账号重复领取新人券
高频领取同一账号反复触发发券逻辑耗尽预算额度
领券退款用券下单后退款，净赚优惠券实际成本远超预期
设备农场用模拟器/群控批量操作大规模套取预算

10.2 三道防线 ​

我们的风控模块实现了三层检查，按执行顺序依次进行：

mermaid
graph LR
    A[用户请求] --> B{设备关联检查}
    B -->|通过| C{日频次检查}
    B -->|拒绝| X[拒绝发券]
    C -->|通过| D{退款率检查}
    C -->|拒绝| X
    D -->|通过| E[正常发券]
    D -->|高退款率| F[降级: 降低面额]

10.2.1 第一道：同设备关联账号数 ​

python
# src/risk/anti_fraud.py

class AntiFraudChecker:
    def __init__(self, config: CouponConfig | None = None) -> None:
        self.config = config or default_config
        # device_id → set of user_ids
        self._device_accounts: dict[str, set[str]] = defaultdict(set)
        # (user_id, date_str) → 当日已发券数
        self._daily_count: dict[tuple[str, str], int] = defaultdict(int)

检查逻辑：维护 device_id → user_ids 的映射，如果同一设备关联的账号数达到阈值（默认 3），拒绝发券：

python
# 1. 同设备关联账号数检查
self._device_accounts[device_id].add(user_id)
if len(self._device_accounts[device_id]) >= self.config.max_accounts_per_device:
    return False, "device_bindcount_exceeded"

配置项：max_accounts_per_device = 3

10.2.2 第二道：单日领券频次 ​

同一用户每天最多领取一定数量的券：

python
# 2. 单日领券上限检查
today = date.today().isoformat()
key = (user_id, today)
if self._daily_count[key] >= self.config.max_coupons_per_day:
    return False, "daily_coupon_limit"

配置项：max_coupons_per_day = 3

10.2.3 第三道：历史退款率 ​

如果用户历史退款率过高，不直接拒绝，而是降级——允许发券但建议降低面额：

python
# 3. 历史退款率检查
if coupon_history is not None and not coupon_history.empty:
    user_hist = coupon_history[coupon_history["user_id"] == user_id]
    if not user_hist.empty and "refunded" in user_hist.columns:
        refund_rate = user_hist["refunded"].mean()
        if refund_rate >= self.config.refund_rate_threshold:
            return True, "downgrade"

配置项：refund_rate_threshold = 0.5（退款率 ≥ 50% 触发降级）

为什么是降级而不是拒绝？

退款率高不一定是恶意行为，也可能是商品不符合预期。直接拒绝可能误伤正常用户。降级策略是一个折中：允许发券但降低面额，减少损失的同时保留转化机会。

10.2.4 通过检查后计数 ​

三道检查都通过后，更新计数器：

python
# 通过 — 计数+1
self._daily_count[key] += 1
return True, "ok"

10.3 完整的检查函数 ​

python
# src/risk/anti_fraud.py

def check(
    self,
    user_id: str,
    device_id: str,
    ip: str,
    coupon_history: pd.DataFrame | None = None,
) -> tuple[bool, str]:
    """检查单个用户是否允许发券.

    Returns
    -------
    (is_allowed, reason)
        is_allowed=False 表示拒绝；
        reason="downgrade" 表示允许但应降低面额。
    """
    # 1. 同设备关联账号数检查
    self._device_accounts[device_id].add(user_id)
    if len(self._device_accounts[device_id]) >= self.config.max_accounts_per_device:
        return False, "device_bindcount_exceeded"

    # 2. 单日领券上限检查
    today = date.today().isoformat()
    key = (user_id, today)
    if self._daily_count[key] >= self.config.max_coupons_per_day:
        return False, "daily_coupon_limit"

    # 3. 历史退款率检查
    if coupon_history is not None and not coupon_history.empty:
        user_hist = coupon_history[coupon_history["user_id"] == user_id]
        if not user_hist.empty and "refunded" in user_hist.columns:
            refund_rate = user_hist["refunded"].mean()
            if refund_rate >= self.config.refund_rate_threshold:
                return True, "downgrade"

    # 通过 — 计数+1
    self._daily_count[key] += 1
    return True, "ok"

10.4 批量检查 ​

线上服务往往需要批量处理。check_batch 封装了逐行检查的逻辑：

python
def check_batch(self, users: pd.DataFrame) -> pd.DataFrame:
    """批量风控检查."""
    results: list[tuple[bool, str]] = []
    has_refund_rate = "refund_rate" in users.columns

    for row in users.itertuples(index=False):
        hist = None
        if has_refund_rate:
            hist = pd.DataFrame({
                "user_id": [row.user_id],
                "refunded": [row.refund_rate],
            })

        allowed, reason = self.check(
            user_id=row.user_id,
            device_id=row.device_id,
            ip=row.ip,
            coupon_history=hist,
        )
        results.append((allowed, reason))

    out = users.copy()
    out["is_allowed"] = [r[0] for r in results]
    out["reason"] = [r[1] for r in results]
    return out

10.5 风控在整体链路中的位置 ​

风控检查在决策流程中的位置很关键——必须在模型打分之后、实际发券之前：

mermaid
graph TD
    A[用户请求] --> B[AB 分桶]
    B --> C[特征提取]
    C --> D[模型打分]
    D --> E[预算优化]
    E --> F[风控检查]
    F -->|通过| G[发券]
    F -->|降级| H[降低面额后发券]
    F -->|拒绝| I[不发券]

为什么不把风控放在最前面？因为风控需要知道这个用户是否会被选中发券。如果用户本来就不在发券名单中，不需要消耗风控的计算资源。

10.6 测试用例 ​

python
# tests/test_optimizer.py

class TestAntiFraudChecker:
    def test_check_normal_user(self, config):
        checker = AntiFraudChecker(config)
        allowed, reason = checker.check("u_001", "device_001", "1.2.3.4")
        assert allowed is True

    def test_check_device_limit(self, config):
        checker = AntiFraudChecker(config)
        # 同设备注册多个账号
        for i in range(config.max_accounts_per_device):
            checker.check(f"u_{i:03d}", "same_device", "1.2.3.4")

        allowed, reason = checker.check("u_999", "same_device", "1.2.3.4")
        assert allowed is False
        assert "device" in reason.lower()

    def test_check_daily_limit(self, config):
        checker = AntiFraudChecker(config)
        for i in range(config.max_coupons_per_day):
            checker.check("u_001", f"device_{i}", "1.2.3.4")

        allowed, reason = checker.check("u_001", "device_new", "1.2.3.4")
        assert allowed is False
        assert "daily" in reason.lower() or "limit" in reason.lower()

10.7 本章小结 ​

  • 防薅羊毛是优惠券系统的必要组成，三道防线层层过滤：
    • 设备关联：同设备最多 3 个账号
    • 日频次限制：单用户每天最多 3 张券
    • 退款率检查：高退款率用户降级而非拒绝
  • 风控位于模型打分之后、实际发券之前
  • 使用内存 dict 实现状态管理，生产环境可替换为 Redis
💡 思考题
  1. 当前实现使用内存 dict 存储设备关联和日频次。如果部署多台服务器，状态不共享怎么办？用 Redis 实现时需要注意哪些并发问题？
  2. IP 参数在当前实现中没有使用。如果要加入 IP 维度的风控（如同一 IP 注册过多账号），应该设什么阈值？家庭 WiFi 和公司网络的 NAT 会带来什么误判问题？
  3. 退款率阈值 50% 是硬编码的。如何用数据驱动的方式确定这个阈值？（提示：考虑退款率分布的长尾特征）

基于 Uplift Modeling 的电商优惠券智能发放系统

© 2025 coupon-rec