如何高效从CSV文件中查找匹配目标金额的商品组合？

问题背景 #

我正在开发一个Python项目，需要从CSV文件中找出总价恰好等于目标金额的商品组合（每个商品对应名称与价格）。目前用递归和itertools.combinations_with_replacement实现的方案在处理大数据集时效率极低——100条记录都要耗很久甚至数天，完全无法处理10万条规模的数据，求更高效的实现思路和优化建议。

原实现代码 #

import csv
from itertools import combinations_with_replacement

def read_products_from_csv(csv_file):
    products = {}
    with open(csv_file, 'r') as file:
        reader = csv.reader(file)
        next(reader)  # Skip the header row
        for row in reader:
            product, price = row
            products[product] = float(price)
    return products

def find_combinations(products, target_amount):
    for r in range(1, len(products) + 1):
        for combination in combinations_with_replacement(products.items(), r):
            total_price = sum(price for _, price in combination)
            if total_price == target_amount:
                return combination
    return None

def main():
    csv_file = "products.csv"  # Replace with your CSV file path
    target_amount = float(input("Enter the sum amount: "))

    products = read_products_from_csv(csv_file)
    combination = find_combinations(products, target_amount)

    if combination:
        print("Combination found:")
        for product, price in combination:
            print(f"{product}: ${price}")
    else:
        print("No combination found.")

if __name__ == "__main__":
    main()

核心优化思路与实现方案 #

1. 解决浮点精度问题，转整数运算 #

浮点数的精度误差会导致total_price == target_amount的判断出现误判，且整数运算效率远高于浮点运算。可以将所有价格乘以100转换为以分为单位的整数，目标金额做同样转换，彻底避免精度问题。

2. 用动态规划（DP）替代暴力枚举 #

这本质是允许重复选择的子集和问题（完全背包问题），动态规划的时间复杂度为O(n*target)，远优于暴力枚举的组合爆炸级复杂度。

动态规划实现示例

def find_combination_dp(products, target_cents):
    # 转换为(商品名, 价格分)的列表
    items = [(name, int(price * 100)) for name, price in products.items()]
    # dp[i] 存储凑出i分的一个可行组合（商品名列表）
    dp = [None] * (target_cents + 1)
    dp[0] = []
    
    for name, price in items:
        # 完全背包正序遍历，允许重复选同一商品
        for i in range(price, target_cents + 1):
            if dp[i - price] is not None and dp[i] is None:
                dp[i] = dp[i - price] + [name]
                # 找到目标组合后直接返回，无需继续遍历
                if i == target_cents:
                    return [(name, products[name]) for name in dp[i]]
    return None

3. 预处理数据，剪枝无效项 #

• 过滤高价商品：直接移除价格大于目标金额的商品，这类商品不可能出现在有效组合中。
• 去重同价商品：将相同价格的商品归类，减少遍历次数（后续匹配时可按需返回所有同价商品）。
• 排序优化：按价格从小到大排序，有助于提前剪枝无效分支。

4. 优先快速判断单商品匹配 #

用哈希表提前检查是否存在价格恰好等于目标金额的商品，这一步能快速返回结果，避免进入复杂的多商品组合计算。

5. 大规模数据的进阶优化 #

如果数据量达到10万级，可进一步优化：

• 滚动数组优化DP内存：只保留当前状态和上一轮状态，降低内存占用。
• 分治策略：将数据集拆分为两部分，分别计算各自的所有子集和，再在两部分中查找和为目标的组合，适合处理超大目标金额场景。

完整优化后代码 #

import csv

def read_products_from_csv(csv_file):
    products = {}
    with open(csv_file, 'r') as file:
        reader = csv.reader(file)
        next(reader)  # 跳过表头
        for row in reader:
            product, price = row
            products[product] = float(price)
    return products

def preprocess_products(products, target_amount):
    # 过滤价格超过目标的商品
    filtered = {name: price for name, price in products.items() if price <= target_amount}
    # 同价商品归类
    price_to_names = {}
    for name, price in filtered.items():
        if price not in price_to_names:
            price_to_names[price] = []
        price_to_names[price].append(name)
    # 按价格从小到大排序
    return sorted(price_to_names.items(), key=lambda x: x[0])

def find_combination_dp(processed_items, target_amount):
    target_cents = int(target_amount * 100)
    dp = [None] * (target_cents + 1)
    dp[0] = []
    
    for price, names in processed_items:
        price_cents = int(price * 100)
        for i in range(price_cents, target_cents + 1):
            if dp[i - price_cents] is not None and dp[i] is None:
                dp[i] = dp[i - price_cents] + [(names[0], price)]
                if i == target_cents:
                    return dp[i]
    return None

def main():
    csv_file = "products.csv"
    target_amount = float(input("请输入目标金额: "))
    
    products = read_products_from_csv(csv_file)
    processed_items = preprocess_products(products, target_amount)
    
    # 先检查单商品匹配
    for price, names in processed_items:
        if abs(price - target_amount) < 1e-9:
            print("找到组合:")
            print(f"{names[0]}: ${price}")
            return
    
    # 动态规划找多商品组合
    combination = find_combination_dp(processed_items, target_amount)
    
    if combination:
        print("找到组合:")
        for product, price in combination:
            print(f"{product}: ${price}")
    else:
        print("未找到匹配的组合。")

if __name__ == "__main__":
    main()

内容的提问来源于stack exchange，提问作者Dead Beats