检查9500万种排列效率低,求Python性能优化建议
问题描述
需要检查约9500万种排列,现有8个不同长度的列表,每个字符串对应Excel中定义的a-k属性(示例:bcdgj代表属性b、c、d、g、j)。目标是找到每个属性至少出现3次的排列,后续再与表格数据匹配。首版Python脚本如下:
import numpy import itertools for x in itertools.product(['abfhj','bcdgj','fghij','abcj','bdgk','abgi','cdei','cdgi','dgik','aghi','abgh','bfhk'],['cdei','bcdgj','abcgi','abcj','abfj','bdfj','cdgi','bhjk','bdgk','dgik'],['afhk','cdgik','cegik','bdgi','cgij','cdei','bcgi','abgh'],['fhjk','bdgij','cgij','abk','ajk','bdk','cik','cdk','cei','fgj'],['abe','abcf','afh','cdi','afj','cdg','abi','cei','cgk','ceg','cgi'],['cdgi','bcgj','bcgi','bcdg','abfh','bdhi','bdgi','bdk','fhk','bei','beg','fgi','abf','abc','egi'],['bcdgik','cegik','chik','afhj','abcj','abfj'],['ceg','bcfg','cgi','bdg','afj','cgj','fhk','cfk','dgk','bcj']): gear = ''.join(x) count_a = gear.count('a') count_b = gear.count('b') count_c = gear.count('c') count_d = gear.count('d') count_e = gear.count('e') count_f = gear.count('f') count_g = gear.count('g') count_h = gear.count('h') count_i = gear.count('i') count_j = gear.count('j') count_k = gear.count('k') score_a = numpy.clip(count_a, 0, 3) score_b = numpy.clip(count_b, 0, 3) score_c = numpy.clip(count_c, 0, 3) score_d = numpy.clip(count_d, 0, 3) score_e = numpy.clip(count_e, 0, 3) score_f = numpy.clip(count_f, 0, 3) score_g = numpy.clip(count_g, 0, 3) score_h = numpy.clip(count_h, 0, 3) score_i = numpy.clip(count_i, 0, 3) score_j = numpy.clip(count_j, 0, 3) score_k = numpy.clip(count_k, 0, 3) rating = score_a + score_b + score_c + score_d + score_e + score_f + score_g + score_h + score_i + score_j + score_k if rating == 33: print(x) print(rating)
当前脚本可运行,但遍历9500万种排列速度极慢,已将各列表元素从数百个缩减至6-12个,需要性能优化方案。
性能优化建议
1. 预计算属性计数,避免重复计算
原脚本每次组合后都要join字符串再多次调用count,这是极大的性能浪费。提前将每个字符串转换成属性计数数组,后续组合时直接累加数组即可。
示例代码:
# 建立属性到索引的映射(a-k对应0-10) attr_index = {'a':0, 'b':1, 'c':2, 'd':3, 'e':4, 'f':5, 'g':6, 'h':7, 'i':8, 'j':9, 'k':10} def preprocess_list(input_list): processed = [] for s in input_list: counts = [0]*11 for char in s: counts[attr_index[char]] += 1 processed.append( (s, counts) ) # 保留原字符串用于结果输出 return processed # 预处理所有8个列表 list1 = preprocess_list(['abfhj','bcdgj','fghij','abcj','bdgk','abgi','cdei','cdgi','dgik','aghi','abgh','bfhk']) list2 = preprocess_list(['cdei','bcdgj','abcgi','abcj','abfj','bdfj','cdgi','bhjk','bdgk','dgik']) list3 = preprocess_list(['afhk','cdgik','cegik','bdgi','cgij','cdei','bcgi','abgh']) list4 = preprocess_list(['fhjk','bdgij','cgij','abk','ajk','bdk','cik','cdk','cei','fgj']) list5 = preprocess_list(['abe','abcf','afh','cdi','afj','cdg','abi','cei','cgk','ceg','cgi']) list6 = preprocess_list(['cdgi','bcgj','bcgi','bcdg','abfh','bdhi','bdgi','bdk','fhk','bei','beg','fgi','abf','abc','egi']) list7 = preprocess_list(['bcdgik','cegik','chik','afhj','abcj','abfj']) list8 = preprocess_list(['ceg','bcfg','cgi','bdg','afj','cgj','fhk','cfk','dgk','bcj'])
遍历组合时直接累加计数数组,计算rating:
import itertools results = [] for combo in itertools.product(list1, list2, list3, list4, list5, list6, list7, list8): total_counts = [0]*11 for s, cnt in combo: for i in range(11): total_counts[i] += cnt[i] # 计算rating:每个属性最多算3分,总和33表示所有属性都满足>=3 rating = sum(min(c, 3) for c in total_counts) if rating == 33: results.append( tuple(s for s, cnt in combo) ) # 批量输出结果 for res in results: print(res)
2. 提前剪枝,过滤不可能的分支
不要等8个元素全选完再判断,而是逐步构建组合,每选一个元素就更新计数,中途如果发现即使剩下的所有元素都贡献该属性的最大值,也无法达到3次,直接跳过这个分支。
示例递归剪枝实现:
# 先预处理所有列表(同步骤1) processed_lists = [list1, list2, list3, list4, list5, list6, list7, list8] # 预计算每个后续列表中各属性的最大出现次数,用于剪枝判断 max_attr_per_list = [] for idx in range(8): max_counts = [0]*11 for s, cnt in processed_lists[idx]: for i in range(11): if cnt[i] > max_counts[i]: max_counts[i] = cnt[i] max_attr_per_list.append(max_counts) def backtrack(current_idx, current_counts, current_selection): if current_idx == 8: # 确认所有属性都>=3 if all(c >=3 for c in current_counts): results.append(tuple(current_selection)) return remaining_steps = 8 - current_idx - 1 for s, cnt in processed_lists[current_idx]: new_counts = [current_counts[i] + cnt[i] for i in range(11)] # 剪枝判断:每个属性是否还有可能达到3次 valid = True for i in range(11): needed = max(0, 3 - new_counts[i]) # 剩下的步骤最多能贡献的该属性数量 max_possible = remaining_steps * max_attr_per_list[current_idx+1][i] if remaining_steps >0 else 0 if needed > max_possible: valid = False break if valid: backtrack(current_idx+1, new_counts, current_selection + [s]) results = [] backtrack(0, [0]*11, []) # 输出结果 for res in results: print(res)
这种剪枝能提前排除大量无效组合,遍历次数会大幅减少。
3. 移除numpy依赖,用纯Python操作替代
原脚本中numpy.clip完全没必要,单个值的裁剪用min(count,3)效率更高,避免numpy的额外开销。
4. 多进程并行处理
遍历是CPU密集型任务,可以用multiprocessing拆分任务到多个进程,利用多核CPU加速。比如把第一个列表的元素拆分成多个chunk,每个进程处理一个chunk对应的所有组合。
示例代码:
from multiprocessing import Pool # 预处理列表(同步骤1) processed_lists = [list1, list2, list3, list4, list5, list6, list7, list8] def process_chunk(start_item): s0, cnt0 = start_item chunk_results = [] # 遍历剩余7个列表的所有组合 for combo in itertools.product(*processed_lists[1:]): total_counts = cnt0.copy() for s, cnt in combo: for i in range(11): total_counts[i] += cnt[i] if all(c >=3 for c in total_counts): chunk_results.append( (s0,) + tuple(s for s, cnt in combo) ) return chunk_results if __name__ == '__main__': # 根据CPU核心数设置进程数,比如4核就设4 with Pool(processes=4) as pool: all_results = pool.map(process_chunk, processed_lists[0]) # 合并并输出结果 for chunk in all_results: for res in chunk: print(res)
5. 优化输出操作
原脚本每次找到结果就调用print,频繁IO会拖慢速度。建议先把结果收集到列表中,最后批量输出,或者直接写入文件。
内容的提问来源于stack exchange,提问作者atlantiantokra
相关产品推荐
相关产品推荐

