带行数限制的API数据检索优化策略求解

我正在开展一个数据采集项目，需从多个统计API获取大量数据。这类API（尤其是老旧政府API）通常不支持全数据集批量下载，要求每次请求指定变量及对应取值，且多数单请求返回行数存在限制。为此，需寻求符合约束的最优请求构造算法。

示例数据集变量 #

{
   "sex": ["total", "women", "men"],
   "country of birth": ["Norway", "Finland", "Sweden", "Denmark"],
   "year": ["2019", "2020", "2021", "2022", "2023"]
}

每次请求必须包含每个变量的至少一个取值，返回行数为各变量选值数量的乘积。例如某请求参数返回3×2×1=6行数据。

该示例中API单请求行数限制为13（row_limit = 13），全量组合共3×4×5=60行，理论最少请求数约为5次（60/13≈4.615）。

目标 #

开发函数生成最优请求配置字典列表，满足以下约束，同时最小化请求次数：

• 每个请求返回行数不超过给定限制；
• 所有请求共同覆盖全部数据组合；
• 请求间无重复数据行。

解决方案思路与算法方向 #

1. 贪心拆分策略（优先处理高基数变量） #

核心逻辑：优先拆分可选值数量最多的变量，高基数变量的拆分能更灵活地控制单请求行数，让每个请求的返回行数尽可能接近限制值，从而减少总请求次数。

• 步骤：
  1. 按变量可选值数量降序排序；
  2. 计算除当前目标变量外，其他变量的取值数量乘积；
  3. 用row_limit除以该乘积，得到当前变量每个子集的最大可选值数量；
  4. 将该变量的取值拆分为多个子集，每个子集与其他变量的全量取值组合成请求，确保单请求行数不超限。

2. 实现示例（Python） #

def generate_optimal_requests(variables, row_limit):
    # 按变量可选值数量从多到少排序
    sorted_vars = sorted(variables.items(), key=lambda x: len(x[1]), reverse=True)
    var_names, var_values = zip(*sorted_vars)
    
    requests = []
    # 计算除第一个变量外，其余变量的取值数量乘积
    other_vars_product = 1
    for vals in var_values[1:]:
        other_vars_product *= len(vals)
    
    # 确定第一个变量的每个子集大小（保证组合后行数不超限）
    subset_size = row_limit // other_vars_product
    # 若乘积超过row_limit，子集大小设为1
    subset_size = max(subset_size, 1)
    
    # 拆分第一个变量的取值为多个子集
    first_var_subsets = [
        var_values[0][i:i+subset_size] 
        for i in range(0, len(var_values[0]), subset_size)
    ]
    
    # 生成每个请求的配置
    for subset in first_var_subsets:
        request_config = {var_names[0]: subset.copy()}
        # 其他变量取全量取值
        for name, vals in zip(var_names[1:], var_values[1:]):
            request_config[name] = vals.copy()
        requests.append(request_config)
    
    return requests

# 示例调用
if __name__ == "__main__":
    target_variables = {
       "sex": ["total", "women", "men"],
       "country of birth": ["Norway", "Finland", "Sweden", "Denmark"],
       "year": ["2019", "2020", "2021", "2022", "2023"]
    }
    row_limit = 13
    optimal_requests = generate_optimal_requests(target_variables, row_limit)
    
    for idx, req in enumerate(optimal_requests, 1):
        print(f"请求 {idx}:")
        for var, vals in req.items():
            print(f"  {var}: {vals}")
        # 计算当前请求的返回行数
        row_count = 1
        for v in req.values():
            row_count *= len(v)
        print(f"  返回行数: {row_count}\n")

运行上述代码会生成5个请求，每个请求返回12行数据（3×4×1=12），刚好覆盖全量60行，达到理论最小请求次数。

3. 进阶算法参考方向 #

• 集合划分变种问题：该问题本质是将多集合笛卡尔积划分为若干大小不超限的子集，属于集合划分的扩展场景，可参考集合划分的优化算法；
• 动态规划优化：对于多变量、高基数的复杂场景，可通过动态规划记录已拆分的变量组合状态，逐步推导最优拆分方案；
• 整数规划模型：当需要严格最优解时，可构建整数规划模型，以最小化请求次数为目标，约束条件为每个请求的行数不超限、覆盖全量组合。

内容的提问来源于stack exchange，提问作者Ruben Selander