第 8 章预算约束优化

对应源码: src/optimizer/budget_optimizer.py
前几章解决了"给谁发"和"发多少面额"的问题，本章要解决最后一个关键约束——在日预算 100 万元的硬约束下，如何选出最优的用户×面额组合，使增量 GMV 最大化。

8.1 从直觉到数学：为什么需要预算优化

假设我们已经有了每个用户的 uplift 分数、预测 GMV 和价格敏感度。一个朴素的想法是：按 uplift 从高到低排序，依次发券直到预算耗尽。

这有什么问题？

一个反直觉的例子

用户 A：uplift=0.12，预测 GMV=50 元，最优面额 30 元
用户 B：uplift=0.08，预测 GMV=500 元，最优面额 5 元

按 uplift 排序，A 排在 B 前面。但 B 的投入产出比远高于 A——花 5 元撬动 500 元 GMV vs 花 30 元撬动 50 元 GMV。

正确做法是看边际效率：每花 1 元预算能带来多少增量收益。

这本质上是一个经典的背包问题。

8.2 背包问题建模

8.2.1 问题的来源

背包问题（Knapsack Problem）是组合优化的经典问题之一：

有一个容量为 $B$ 的背包和 $n$ 个物品。物品 $i$ 的重量为 $w_{i}$ 、价值为 $v_{i}$ 。如何选择物品装入背包，使总价值最大、且总重量不超过 $B$ ？

在优惠券场景下的对应关系：

背包概念	优惠券含义
背包容量 $B$	日预算 100 万元
物品 $i$	候选用户 $i$
重量 $w_{i}$	期望发券成本
价值 $v_{i}$	期望增量 GMV × 毛利率

8.2.2 为什么是分组背包

标准 0-1 背包每个物品只有"放 / 不放"两种选择。但我们的情况更复杂——每个用户有多个可选面额 $(5, 10, 15, 20, 30, 50)$ 元，选不同面额对应不同的成本和收益。

这是分组背包（Group Knapsack）：

max \sum_{i = 1}^{n} {gain}_{i} (d_{i}) s.t. \sum_{i = 1}^{n} {cost}_{i} (d_{i}) \leq B

其中 $d_{i} \in {0, 5, 10, 15, 20, 30, 50}$ 是给用户 $i$ 分配的面额（ $d_{i} = 0$ 表示不发券）。

8.2.3 面额响应函数

不同面额产生不同的增量转化概率。我们用对数曲线建模这种边际递减效应：

inc_prob (d) = uplift \times \frac{\log (1 + d \cdot s)}{\log (1 + d_{ref} \cdot s)}

其中：

$s = 0.5 + sensitivity$ ，范围 $[0.5, 1.5]$
$d_{ref}$ 是面额集合中位数（15 元），使中位面额时因子 ≈ 1

直觉：高敏感度用户（ $s$ 大）对大面额的响应更强；低敏感度用户面额增大后边际效应递减更快。

python

# src/optimizer/budget_optimizer.py

def _amount_response_factor(
    self, base_uplift: float, amount: int, sensitivity: float
) -> float:
    """面额响应函数 — 对数曲线调整."""
    if base_uplift <= 0 or amount <= 0:
        return 0.0

    sensitivity = max(0.0, min(1.0, sensitivity))
    ref_amount = float(np.median(self.config.coupon_amounts))
    scale = 0.5 + sensitivity  # 范围 [0.5, 1.5]

    numerator = math.log1p(amount * scale)
    denominator = math.log1p(ref_amount * scale)
    if denominator == 0:
        return base_uplift

    return base_uplift * (numerator / denominator)

有了这个函数，每个 (用户, 面额) 组合的期望成本和收益就能计算出来：

{cost}_{i} (d) = d \times {inc_prob}_{i} (d) {gain}_{i} (d) = {inc_prob}_{i} (d) \times {GMV}_{i} \times margin

8.3 策略一：贪心求解

8.3.1 核心思想

分组背包是 NP-hard 的，动态规划在百万用户×6 种面额下不可行。贪心算法是工业界最常用的近似方案：

对每个用户，枚举所有面额，选出效率最高的面额（efficiency = gain / cost）
将所有用户按效率降序排序
从高到低逐个分配，直到预算耗尽

时间复杂度 $O (n \cdot | D | + n \log n)$ ，其中 $| D |$ 是面额数。

8.3.2 实现

python

# src/optimizer/budget_optimizer.py

def optimize(self, candidates: pd.DataFrame) -> pd.DataFrame:
    """贪心策略: 按边际效率降序分配直到预算耗尽."""
    self._validate_candidates(candidates)
    pool = self._build_best_per_user(candidates)
    if pool.empty:
        return pd.DataFrame(columns=_RESULT_COLUMNS)

    pool.sort_values("efficiency", ascending=False, inplace=True)

    budget_left = self.config.daily_budget
    keep_idx: list[int] = []
    seen: set[str] = set()
    for idx, r in pool.iterrows():
        uid = r["user_id"]
        if uid in seen:
            continue
        cost = r["expected_cost"]
        if cost > budget_left:
            continue
        keep_idx.append(idx)
        seen.add(uid)
        budget_left -= cost

    return pool.loc[keep_idx].reset_index(drop=True)

_build_best_per_user 为每个用户选出效率最高的面额：

python

def _build_best_per_user(self, candidates: pd.DataFrame) -> pd.DataFrame:
    """为每个用户选出效率最高的面额 (贪心用)."""
    records: list[dict] = []
    for row in candidates.itertuples(index=False):
        best: dict | None = None
        for amount in self.config.coupon_amounts:
            inc_prob = self._amount_response_factor(
                row.uplift_score, amount, row.sensitivity_score
            )
            if inc_prob <= 0:
                continue
            exp_cost = amount * inc_prob
            exp_gain = inc_prob * row.predicted_gmv * self.config.margin_rate
            if exp_cost <= 0:
                continue
            eff = exp_gain / exp_cost
            if best is None or eff > best["efficiency"]:
                best = {
                    "user_id": row.user_id,
                    "coupon_amount": amount,
                    "expected_cost": exp_cost,
                    "expected_gain": exp_gain,
                    "efficiency": eff,
                }
        if best is not None:
            records.append(best)

    if not records:
        return pd.DataFrame(columns=_RESULT_COLUMNS)
    return pd.DataFrame(records)

8.3.3 贪心的局限

贪心的最大问题是面额选择与预算分配耦合不够好：

先为每个用户选"效率最高"的面额（可能都选了 5 元），再排序裁剪
如果预算宽松，给高价值用户提高面额（效率稍低但绝对收益更高）可能更优
贪心无法做到在"多发 10 块给用户 A"和"少发 5 块给用户 B"之间全局权衡

这就是为什么我们需要拉格朗日松弛。

8.4 策略二：拉格朗日松弛

8.4.1 从有约束到无约束

拉格朗日松弛的核心思想是用一个影子价格 $λ$ 把约束融入目标函数。

原问题（有约束）：

max \sum_{i} {gain}_{i} (d_{i}) s.t. \sum_{i} {cost}_{i} (d_{i}) \leq B

引入拉格朗日乘子 $λ \geq 0$ ：

L (λ) = max \sum_{i} [{gain}_{i} (d_{i}) - λ \cdot {cost}_{i} (d_{i})] + λ B

为什么这样做？

每花 1 元预算，需要"付出" $λ$ 的代价
$λ$ 就是每元预算的影子价格——预算有多紧张
每个用户的决策变成：选面额使 $gain (d) - λ \cdot cost (d)$ 最大
各用户的子问题完全独立，可以并行求解

8.4.2 二分搜索影子价格

$λ$ 的值决定了"预算紧张程度"：

$λ = 0$ ：预算免费，每个用户选收益最大的面额 → 总成本最大
$λ \to \infty$ ：预算极贵，所有人都不发券 → 总成本 = 0

所以 $λ$ 与总成本是单调递减关系，可以用二分搜索找到使总成本 ≈ 预算 $B$ 的 $λ^{*}$ ：

lam_lo = 0, lam_hi = max(gain/cost) × 2
while |total_cost - B| / B > tolerance:
    lam = (lam_lo + lam_hi) / 2
    对每个用户，选 argmax[gain(d) - lam × cost(d)]
    total_cost = Σ cost_i(d_i*)
    if total_cost > B:
        lam_lo = lam    # 预算太便宜，涨价
    else:
        lam_hi = lam    # 预算太贵，降价

8.4.3 完整实现

python

# src/optimizer/budget_optimizer.py

def optimize_lagrangian(
    self,
    candidates: pd.DataFrame,
    *,
    max_iter: int = 80,
    tol: float = 0.005,
) -> LagrangianResult:
    """拉格朗日松弛策略: 二分搜索影子价格 λ，使总成本逼近预算."""
    self._validate_candidates(candidates)
    budget = self.config.daily_budget

    # 预计算每个用户在每个面额下的 (cost, gain, amount)
    user_options = self._precompute_user_options(candidates)

    if not user_options:
        return LagrangianResult(
            allocation=pd.DataFrame(columns=_RESULT_COLUMNS),
            lam=0.0, total_cost=0.0, total_gain=0.0, iterations=0,
        )

    # 确定 λ 的搜索范围
    lam_lo = 0.0
    lam_hi = 0.0
    for opts in user_options.values():
        for cost, gain, _amount in opts:
            if cost > 0:
                lam_hi = max(lam_hi, gain / cost)
    lam_hi *= 2.0  # 留余量

    # 二分搜索
    best_lam = 0.0
    best_alloc: dict[str, tuple[int, float, float]] = {}
    iterations = 0

    for i in range(max_iter):
        iterations = i + 1
        lam = (lam_lo + lam_hi) / 2.0

        alloc = self._solve_subproblem(user_options, lam)
        total_cost = sum(cost for _, cost, _ in alloc.values())

        if abs(total_cost - budget) / max(budget, 1e-9) < tol:
            best_lam = lam
            best_alloc = alloc
            break

        if total_cost > budget:
            lam_lo = lam    # 超预算 → λ 太小，提高 λ
        else:
            lam_hi = lam    # 欠预算 → λ 太大，降低 λ

        best_lam = lam
        best_alloc = alloc

    # 后处理: 超预算时按净收益排序裁剪
    total_cost = sum(cost for _, cost, _ in best_alloc.values())
    if total_cost > budget:
        best_alloc = self._trim_to_budget(best_alloc, budget, best_lam)
        total_cost = sum(cost for _, cost, _ in best_alloc.values())

    # 构建结果
    records = []
    for uid, (amount, cost, gain) in best_alloc.items():
        if amount > 0:
            eff = gain / cost if cost > 0 else 0.0
            records.append({
                "user_id": uid,
                "coupon_amount": amount,
                "expected_cost": cost,
                "expected_gain": gain,
                "efficiency": eff,
            })

    alloc_df = pd.DataFrame(records, columns=_RESULT_COLUMNS)
    total_gain = alloc_df["expected_gain"].sum() if not alloc_df.empty else 0.0

    return LagrangianResult(
        allocation=alloc_df, lam=best_lam,
        total_cost=total_cost, total_gain=total_gain,
        iterations=iterations,
    )

8.4.4 子问题求解

给定 $λ$ ，每个用户独立选择最优面额：

python

def _solve_subproblem(
    self,
    user_options: dict[str, list[tuple[float, float, int]]],
    lam: float,
) -> dict[str, tuple[int, float, float]]:
    """给定 λ，每个用户独立选使 (gain - λ×cost) 最大的面额."""
    alloc: dict[str, tuple[int, float, float]] = {}

    for uid, opts in user_options.items():
        best_net = 0.0  # 不发券的净收益为 0
        best_amount = 0
        best_cost = 0.0
        best_gain = 0.0

        for cost, gain, amount in opts:
            net = gain - lam * cost
            if net > best_net:
                best_net = net
                best_amount = amount
                best_cost = cost
                best_gain = gain

        if best_amount > 0:
            alloc[uid] = (best_amount, best_cost, best_gain)

    return alloc

注意初始值

best_net = 0.0 是关键——这代表"不发券"的收益为 0。只有当某个面额的净收益 $gain - λ \cdot cost > 0$ 时，才会选择发券。这自动过滤了不划算的用户。

8.4.5 超预算裁剪

二分搜索结束后，可能仍有轻微超预算（因为 $λ$ 是连续的但面额是离散的）。此时按净收益排序，移除收益最低的分配：

python

def _trim_to_budget(
    self,
    alloc: dict[str, tuple[int, float, float]],
    budget: float,
    lam: float,
) -> dict[str, tuple[int, float, float]]:
    """超预算时，按净收益 (gain - λ×cost) 排序，移除最低的."""
    items = [
        (uid, amount, cost, gain, gain - lam * cost)
        for uid, (amount, cost, gain) in alloc.items()
    ]
    items.sort(key=lambda x: x[4], reverse=True)

    trimmed: dict[str, tuple[int, float, float]] = {}
    budget_left = budget
    for uid, amount, cost, gain, _ in items:
        if cost <= budget_left:
            trimmed[uid] = (amount, cost, gain)
            budget_left -= cost

    return trimmed

8.5 两种策略对比

维度	贪心	拉格朗日松弛
时间复杂度	$O(n	D
面额选择	先选效率最高，再排序	面额选择与预算分配联合优化
可并行	排序后顺序分配	子问题完全独立，可并行
预算利用率	一般	高（逼近预算上限）
约束扩展	困难	容易加入多维约束
实际差距	中小规模接近	大规模/预算紧张时优势明显

工业实践

美团/京东的大促预算分配通常使用拉格朗日松弛或其变体（如 ADMM）。原因：

百万级用户的子问题可以用 MapReduce 并行
容易扩展到多维约束（预算 + 券量 + 品类均衡）
$λ$ 有经济学含义——"影子价格"就是预算的边际价值

8.6 测试验证

python

# tests/test_optimizer.py

class TestBudgetOptimizer:
    def test_optimize_respects_budget(self, config, sample_candidates):
        optimizer = BudgetOptimizer(config)
        result = optimizer.optimize(sample_candidates)

        assert isinstance(result, pd.DataFrame)
        if "expected_cost" in result.columns and not result.empty:
            assert result["expected_cost"].sum() <= config.daily_budget * 1.01

    def test_optimize_only_positive_uplift(self, config, sample_candidates):
        optimizer = BudgetOptimizer(config)
        result = optimizer.optimize(sample_candidates)

        if not result.empty:
            merged = result.merge(sample_candidates, on="user_id")
            assert (merged["uplift_score"] > 0).all()


class TestLagrangianOptimizer:
    def test_lagrangian_respects_budget(self, config, sample_candidates):
        optimizer = BudgetOptimizer(config)
        result = optimizer.optimize_lagrangian(sample_candidates)

        assert isinstance(result, LagrangianResult)
        assert result.total_cost <= config.daily_budget * 1.01

    def test_lagrangian_tight_budget_uses_budget(self):
        """预算紧张时，拉格朗日应尽量用完预算."""
        config = CouponConfig(daily_budget=50)  # 极小预算
        rng = np.random.default_rng(42)
        candidates = pd.DataFrame({
            "user_id": [f"u_{i:04d}" for i in range(100)],
            "uplift_score": rng.uniform(0.05, 0.20, size=100),
            "predicted_gmv": rng.lognormal(4.5, 0.3, size=100),
            "sensitivity_score": rng.beta(2, 5, size=100),
        })
        optimizer = BudgetOptimizer(config)
        result = optimizer.optimize_lagrangian(candidates)

        if result.total_cost > 0:
            utilization = result.total_cost / config.daily_budget
            assert utilization > 0.5  # 预算利用率 > 50%

关键测试点：

预算硬约束：总成本不超过预算（允许 1% 浮点容差）
只对正 uplift 用户发券：负 uplift 用户不会被选中
预算利用率：在紧张预算下，拉格朗日应尽量用完预算
两种方法可比性：相同输入下，贪心和拉格朗日的总收益应在同一量级

8.7 本章小结

优惠券预算分配本质上是一个分组背包问题
面额响应函数用对数曲线建模面额-转化概率的边际递减关系
贪心策略简单高效，按边际效率排序，适合快速决策
拉格朗日松弛将预算约束转化为影子价格 $λ$ ，通过二分搜索找到最优 $λ^{*}$ ，使面额选择与预算分配联合优化
拉格朗日方法的子问题完全独立、可并行，是工业界大规模预算分配的主流方案

💡 思考题

如果除了总预算约束外，还要限制"每种面额最多发 N 张"（面额配额约束），应该如何扩展拉格朗日松弛？需要引入几个 $λ$ ？
面额响应函数使用了对数曲线，如果用 S 型曲线（sigmoid）替代会有什么区别？什么场景下 sigmoid 更合适？
贪心方法先为每个用户选"效率最高"的面额。能否设计一种"延迟决策"的贪心：先排序用户，再根据剩余预算动态选面额？这和拉格朗日方法有什么关系？

第 8 章 预算约束优化 ​

8.1 从直觉到数学：为什么需要预算优化 ​

8.2 背包问题建模 ​

8.2.1 问题的来源 ​

8.2.2 为什么是分组背包 ​

8.2.3 面额响应函数 ​

8.3 策略一：贪心求解 ​

8.3.1 核心思想 ​

8.3.2 实现 ​

8.3.3 贪心的局限 ​

8.4 策略二：拉格朗日松弛 ​

8.4.1 从有约束到无约束 ​

8.4.2 二分搜索影子价格 ​

8.4.3 完整实现 ​

8.4.4 子问题求解 ​

8.4.5 超预算裁剪 ​

8.5 两种策略对比 ​

8.6 测试验证 ​

8.7 本章小结 ​

第 8 章预算约束优化

8.1 从直觉到数学：为什么需要预算优化

8.2 背包问题建模

8.2.1 问题的来源

8.2.2 为什么是分组背包

8.2.3 面额响应函数

8.3 策略一：贪心求解

8.3.1 核心思想

8.3.2 实现

8.3.3 贪心的局限

8.4 策略二：拉格朗日松弛

8.4.1 从有约束到无约束

8.4.2 二分搜索影子价格

8.4.3 完整实现

8.4.4 子问题求解

8.4.5 超预算裁剪

8.5 两种策略对比

8.6 测试验证

8.7 本章小结