Python替代PULP求解大规模数据集多目标约束选行优化问题
千万级数据集下的约束优化筛选方案
针对千万级规模的DataFrame,传统整数规划工具(比如PULP)因变量数量过大导致计算超时,这里提供一套预筛选+精确/近似优化的解决方案,兼顾速度与最优性:
核心思路
先通过单行目标得分筛选出高价值候选行,把千万级数据压缩到几百至几千行的可控规模,再用整数规划或贪心算法完成带约束的最优选择。
步骤1:预筛选缩小候选池
- 给每行计算目标得分:
score = Rev + GM + RM,这是单行对总目标的贡献值。 - 针对每个(产品+促销类别)组合(共9组:P1_Base、P1_HP、P1_Deep、P2_Base...P3_Deep),分别按
score降序取前N行(N要大于约束要求的最小行数,比如P1_Base取前50行,确保后续有足够候选满足约束)。 - 合并所有筛选后的行并去重(一行可能同时属于多个组),得到小规模候选池。
预筛选代码示例:
import pandas as pd # 假设原始数据为df,包含Rev、GM、RM、P1_Promo、P2_Promo、P3_Promo列 df['score'] = df['Rev'] + df['GM'] + df['RM'] # 按每个产品的促销类别筛选高得分行 group_configs = [ ('P1_Promo', 'Base', 50), ('P1_Promo', 'HP', 20), ('P1_Promo', 'Deep', 20), ('P2_Promo', 'Base', 60), ('P2_Promo', 'HP', 20), ('P2_Promo', 'Deep', 20), ('P3_Promo', 'Base', 50), ('P3_Promo', 'HP', 20), ('P3_Promo', 'Deep', 20), ] candidate_dfs = [] for col, promo_type, top_n in group_configs: top_rows = df[df[col] == promo_type].nlargest(top_n, 'score') candidate_dfs.append(top_rows) # 合并去重,得到候选池 candidates = pd.concat(candidate_dfs).drop_duplicates().reset_index(drop=True)
步骤2:在候选池上做精确整数规划
候选池规模缩小后,用更高效的整数规划库(比如Google OR-Tools)快速求解最优解,比PULP性能更优。
OR-Tools求解代码示例:
from ortools.linear_solver import pywraplp # 初始化SCIP求解器(适合整数规划问题) solver = pywraplp.Solver.CreateSolver('SCIP') if not solver: raise RuntimeError("无法初始化求解器") # 定义变量:x[i]表示是否选中第i行(0=不选,1=选) n_candidates = len(candidates) x = [solver.IntVar(0, 1, f'x_{i}') for i in range(n_candidates)] # 目标函数:最大化总得分 solver.Maximize(solver.Sum([x[i] * candidates['score'].iloc[i] for i in range(n_candidates)])) # 总约束:必须选52行 solver.Add(solver.Sum(x) == 52) # P1促销类别约束 solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P1_Promo'].iloc[i] == 'Base']) >= 37) solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P1_Promo'].iloc[i] == 'HP']) >= 5) solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P1_Promo'].iloc[i] == 'Deep']) >= 10) # P2促销类别约束 solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P2_Promo'].iloc[i] == 'Base']) >= 40) solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P2_Promo'].iloc[i] == 'HP']) >= 5) solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P2_Promo'].iloc[i] == 'Deep']) >= 7) # P3促销类别约束 solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P3_Promo'].iloc[i] == 'Base']) >= 37) solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P3_Promo'].iloc[i] == 'HP']) >= 7) solver.Add(solver.Sum([x[i] for i in range(n_candidates) if candidates['P3_Promo'].iloc[i] == 'Deep']) >= 8) # 求解并提取结果 status = solver.Solve() if status == pywraplp.Solver.OPTIMAL: selected_indices = [i for i in range(n_candidates) if x[i].solution_value() == 1] selected_rows = candidates.iloc[selected_indices] print("找到最优解,总得分:", solver.Objective().Value()) else: print("未找到可行解,请检查约束或扩大候选池规模")
步骤3:近似解(极速方案)
如果不需要绝对最优,只想快速得到满足约束的解,可以用贪心算法:
- 初始化各约束的已满足计数为0,已选行数为0,候选池复制为临时池。
- 循环直到选够52行:
- 给每行计算优先级:
score + 1000 * 未满足约束中该行能覆盖的数量(加权值1000确保优先选能补约束的行) - 选中优先级最高的行,更新对应约束的计数和已选行数,从临时池中移除该行。
- 给每行计算优先级:
- 最后检查所有约束是否满足,若有未满足的,替换部分低score的行,换成能补约束的高score行。
方案优势
- 预筛选用Pandas矢量化操作,处理千万级数据仅需几秒到几十秒,效率极高。
- 后续优化在小规模候选池上运行,整数规划几秒出结果,贪心算法毫秒级完成。
- 兼顾最优性与速度,完全适配千万级数据集的需求。
内容的提问来源于stack exchange,提问作者Manny
相关产品推荐
相关产品推荐

