如何用Pulp解决短信渠道成本优化中的阶梯定价约束
用PuLP实现短信渠道阶梯式定价的线性规划优化
我们正在做短信渠道发送成本优化,手上有数百个分属不同供应商、运营商的渠道,每个渠道支持的短信类型、定价规则、日发送限额都不一样。目前已经把基本约束(比如日发送量限制)放进了线性规划模型,但卡在阶梯式定价的实现上——比如某渠道发送量在0-1000条时单价0.03元,1001-5000条时单价0.025元,5000条以上单价0.02元。当前代码只硬编码了第一阶梯的单价,明显不符合实际成本规则,原代码如下:
##TODO minimum price expression sentPrice=None SHlinkNum=None SZMoveNum=None for dictType in channel_vars: for typeKeys in dictType.keys(): if supplier["SHlink"].__contains__(typeKeys): SHlinkNum+=dictType[typeKeys] elif supplier["SZmove"].__contains__(typeKeys): SZMoveNum+=dictType[typeKeys] chanPrice=round(cost.get(typeKeys,0),2) sentPrice+=chanPrice*dictType[typeKeys] SHsuppNum=supplierSent.get("SHlink",0) SZsuppNum=supplierSent.get("SZmove",0) ##TODO Channel Minimum Cost Constraints prob+=SHlinkNum==int((supplierTarget["SHlink"]-SHsuppNum)/date_diff),"SHlinkLimitNum" prob+=SZMoveNum==int((supplierTarget["SZmove"]-SZsuppNum)/date_diff),"SZmoveLimitNum" prob+=(lpSum(sentPrice+SHsuppNum*0.0+SZMoveNum*0.0), "Total cost of channels per chan" )
解决方案:拆分阶梯变量实现分段成本计算
线性规划无法直接处理非线性的阶梯成本,核心思路是把单一的发送量变量拆分成对应每个阶梯的分段变量,每个变量绑定对应阶梯的单价,再通过约束确保分段变量的总和等于实际发送量,同时限制每个分段变量不超过对应阶梯的容量(最后一个阶梯无上限)。
修改后的代码实现
from pulp import LpProblem, LpMinimize, LpVariable, lpSum # 假设已初始化规划问题:prob = LpProblem("SMS_Cost_Optimization", LpMinimize) # 假设channel_vars是各渠道的发送量变量集合,每个dictType为{短信类型: LpVariable} # 定义阶梯成本结构(可从配置/数据库读取,适配数百个渠道) tiered_costs = { "SHlink_typeA": [(1000, 0.03), (4000, 0.025), (float('inf'), 0.02)], # 0-1000:0.03, 1001-5000:0.025, 5000+:0.02 "SZmove_typeB": [(2000, 0.032), (3000, 0.028), (float('inf'), 0.024)], # 其他短信类型的阶梯规则... } total_cost = 0 SHlinkNum = 0 SZMoveNum = 0 for dictType in channel_vars: for typeKey, send_var in dictType.items(): # 累加供应商总发送量(保留原逻辑) if typeKey in supplier["SHlink"]: SHlinkNum += send_var elif typeKey in supplier["SZmove"]: SZMoveNum += send_var # 处理阶梯成本 tiers = tiered_costs.get(typeKey, [(float('inf'), round(cost.get(typeKey, 0), 2))]) # 默认单阶梯兼容原逻辑 prev_limit = 0 tier_vars = [] for i, (limit, price) in enumerate(tiers): # 创建对应阶梯的变量 tier_var = LpVariable(f"{typeKey}_tier{i}", lowBound=0, cat='Continuous') tier_vars.append(tier_var) # 约束:当前阶梯变量不超过该阶梯的最大容量(最后一个阶梯无上限) if limit != float('inf'): prob += tier_var <= (limit - prev_limit), f"{typeKey}_tier{i}_cap" # 累加该阶梯的成本 total_cost += price * tier_var prev_limit = limit # 约束:所有阶梯变量总和等于该渠道总发送量 prob += lpSum(tier_vars) == send_var, f"{typeKey}_total_send" # 保留原供应商发送量约束 SHsuppNum = supplierSent.get("SHlink", 0) SZsuppNum = supplierSent.get("SZmove", 0) prob += SHlinkNum == int((supplierTarget["SHlink"] - SHsuppNum)/date_diff), "SHlinkLimitNum" prob += SZMoveNum == int((supplierTarget["SZmove"] - SZsuppNum)/date_diff), "SZmoveLimitNum" # 设置目标函数:最小化总成本 prob += total_cost, "Total_Cost" # 求解模型 prob.solve()
关键说明
- 如果发送量要求整数,把
tier_var的cat='Continuous'改成cat='Integer',但会增加求解时间。 - 阶梯成本结构可通过配置文件或数据库动态加载,无需硬编码,适配大规模渠道场景。
- 原代码中直接用单价乘总量的简单计算被替换为阶梯变量的分段求和,精准匹配阶梯定价规则。
内容的提问来源于stack exchange,提问作者user21040411
相关产品推荐
相关产品推荐

