广告投放问题¶

类别： 非线性优化

使用 OptAgent 的 Python 接口描述变量、约束与目标。

问题与原始示例来源：Hexaly Code Templates。

问题描述¶

在广告投放问题中,一组广告可以在各个时间段上播出。每条广告在触达目标受众(例如特定年龄段)时会产生预期的营销收益。当一条广告被分配到某个时间段时,它会以给定的概率触达其目标受众。如果该广告在多个时间段播出,该概率会增大。目标是为每个时段分配一条广告,以使整个广告活动的总预期收益最大化。

该问题也被称为武器目标分配问题(Weapon Target Assignment Problem, WTA)。关于该问题的更多细节,请参阅维基百科。

学习要点¶

使用 OptAgent 的 set 建模分配给每个广告的时段
使用 array 和 partition 确保每个时段恰好分配给一条广告
使用 prod 和 lambda_function 构建每条广告的非线性预期收益

数据¶

本算例是对 GitHub 上提供的武器目标分配算例的略微修改版本。数据格式如下:

第一行:时段数量,广告数量
接下来每一行对应一条广告的营销收益
之后的每一行包含:
时段的索引
广告的索引
该广告在该时段触达目标受众的概率

建模思路¶

广告投放问题的 OptAgent 模型使用集合决策变量。我们通过 set 为每条广告定义一个集合,其中包含分配给该广告的所有时段；再用 array 将这些集合组成一等数组,并施加 partition 约束,确保每个时段恰好被分配一条广告。

对于每条广告,我们用 lambda_function 表达各个已分配时段的未触达概率,再用集合聚合 prod 计算整场活动均未触达的概率。用 1 减去该乘积并乘以营销收益,即可得到广告的预期收益。最后对所有广告的预期收益求和并最大化。

Python 实现¶

from pathlib import Path

from optagent import OptModel, solve


def read_instance(instance_filename):
    with open(instance_filename, encoding="utf-8") as f:
        lines = f.readlines()

    nb_slots = int(lines[0].split()[0])
    nb_ads = int(lines[0].split()[1])
    ad_profits = [int(lines[i]) for i in range(1, nb_ads + 1)]
    probabilities_data = [[0.0 for _ in range(nb_ads)] for _ in range(nb_slots)]
    for line in lines[nb_ads + 1 :]:
        values = line.split()
        s = int(values[0])
        a = int(values[1])
        probabilities_data[s][a] = float(values[2])
    return nb_slots, nb_ads, ad_profits, probabilities_data


def main(input_file, output_file=None, time_limit=60):
    _, _, ad_profits, probabilities_data = read_instance(input_file)
    nb_slots = len(probabilities_data)
    nb_ads = len(ad_profits)
    model = OptModel()

    # assignments[a] is the set of slots assigned to ad a.
    assignments = [model.set(nb_slots) for a in range(nb_ads)]
    assignments_array = model.array(assignments)

    # Every slot must be assigned to exactly one ad.
    model.constraint(model.partition(assignments_array))

    # The empty product is 1, so an ad with no slots contributes zero profit.
    probabilities = model.array(probabilities_data)
    ad_profit_expressions = []
    for a in range(nb_ads):
        ad_index = a
        miss_probability = model.prod(
            assignments[a],
            model.lambda_function(
                lambda slot: 1 - model.at(probabilities, slot // 1, ad_index)
            ),
        )
        ad_profit_expressions.append(ad_profits[a] * (1 - miss_probability))
    total_profit = model.sum(ad_profit_expressions)

    model.maximize(total_profit)
    solution = solve(model, time_limit_s=float(time_limit))
    result_values = {'total_profit': total_profit.value, **{f'ad_{a}': assignment.value for a, assignment in enumerate(assignments)}}

    lines = [
        f"Nb slots = {nb_slots}; Nb ads = {nb_ads}; "
        f"Total profit = {result_values['total_profit']}"
    ]
    for a in range(nb_ads):
        assigned_slots = result_values[f"ad_{a}"]
        if assigned_slots:
            slots = ", ".join(f"slot {slot}" for slot in sorted(assigned_slots))
            lines.append(f"Ad {a} assigned to {slots}")

    result_text = "\n".join(lines)
    print(result_text)
    if output_file is not None:
        Path(output_file).write_text(result_text + "\n", encoding="utf-8")
    return solution

本地运行¶

Notebook 直接调用 main 并显式传入实例路径。以下代码格相互独立，可以按需要单独运行；调整 time_limit 可以控制每个实例的求解时间。

from pathlib import Path

INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)

solution_50x100 = main(
    INSTANCE_DIR / "ad_campaign_50x100.txt",
    time_limit=1,
)

solution_100x200 = main(
    INSTANCE_DIR / "ad_campaign_100x200.txt",
    time_limit=1,
)

solution_150x300 = main(
    INSTANCE_DIR / "ad_campaign_150x300.txt",
    time_limit=1,
)