当前位置：首页 > news >正文

Pythonitertools高级模式

news 2026/6/3 11:33:17

Python itertools 高级模式
===============================

itertools 提供高效的迭代器工具, 所有函数均返回迭代器——惰性求值, 内存友好.

1. 无限迭代器: cycle / islice / tee
-------------------------------------

from itertools import cycle, islice, tee

# cycle 无限循环可迭代对象
counter = 0
for item in cycle(["红", "黄", "绿"]):
print("循环信号灯:", item)
counter += 1
if counter >= 5:
break

# islice 对迭代器做切片 (惰性)
# islice(iterable, start, stop, step)
data = range(100)
subset = islice(data, 10, 30, 2) # 索引10到30, 步长2
print("islice 切片结果:", list(subset))

# tee 将一个迭代器克隆为 N 个独立副本
src = range(5)
a, b, c = tee(src, 3) # 克隆出3个独立迭代器
print("tee 迭代器a:", list(a)) # [0,1,2,3,4]
print("tee 迭代器b:", list(b)) # [0,1,2,3,4]
print("tee 迭代器c:", list(c)) # [0,1,2,3,4]
# 注意: tee 会缓存数据, 如果原始迭代器很大且多个副本进度差异大, 会消耗内存

2. 组合数学: combinations / permutations / product
-----------------------------------------------------

from itertools import combinations, permutations, product

# combinations: 无放回组合, 元素不重复, 顺序无关
items = ["A", "B", "C"]
print("组合(2个):", list(combinations(items, 2)))
# [('A','B'), ('A','C'), ('B','C')]

# permutations: 排列, 顺序有关, 元素不重复
print("排列(2个):", list(permutations(items, 2)))
# [('A','B'), ('A','C'), ('B','A'), ('B','C'), ('C','A'), ('C','B')]

# product: 笛卡尔积, 相当于嵌套循环, repeat 参数允许重复
print("笛卡尔积:", list(product([0, 1], repeat=2)))
# [(0,0), (0,1), (1,0), (1,1)]

# 实战: 枚举所有可能的密码组合
digits = "0123456789"
# product(digits, repeat=4) 生成所有4位数字密码
for pin in islice(product(digits, repeat=4), 3):
print("密码组合:", "".join(pin))

3. groupby — 分组聚合
-----------------------
groupby 要求输入已排序, 只对连续相同键分组.

from itertools import groupby

data = [("北京", 100), ("北京", 200), ("上海", 150), ("上海", 300), ("广州", 80)]
# 按城市分组前必须先排序
sorted_data = sorted(data, key=lambda x: x[0])

result = {}
for city, group in groupby(sorted_data, key=lambda x: x[0]):
values = [item[1] for item in group]
result[city] = {"总量": sum(values), "次数": len(values), "平均": sum(values)/len(values)}
print("groupby 分组统计:", result)

# groupby 返回的 group 是迭代器, 需要遍历前消费

4. accumulate — 累计运算
--------------------------

from itertools import accumulate
import operator

nums = [1, 2, 3, 4, 5]

# 默认累加
print("累计求和:", list(accumulate(nums))) # [1, 3, 6, 10, 15]

# 累计乘积
print("累计乘积:", list(accumulate(nums, operator.mul))) # [1, 2, 6, 24, 120]

# 自定义函数: 累计最大值
print("累计最大值:", list(accumulate(nums, max))) # [1, 2, 3, 4, 5]

# 斐波那契数列
def fib():
yield 0
yield 1
yield from accumulate(range(2, 20), lambda a, b: a + b)
# 注意: accumulate 的 func 接收 (上一次结果, 当前元素)

5. chain / chain.from_iterable — 展平
-----------------------------------------

from itertools import chain

# chain 连接多个可迭代对象
lists = [[1, 2], [3, 4], [5]]
flattened = chain.from_iterable(lists) # 等价于 chain(*lists)
print("展平列表:", list(flattened)) # [1, 2, 3, 4, 5]

# chain 可混合不同类型
mixed = chain("ABC", [1, 2], (3,))
print("混合类型连接:", list(mixed)) # ['A','B','C',1,2,3]

6. zip_longest — 最长拉链
--------------------------

from itertools import zip_longest

names = ["Alice", "Bob", "Charlie"]
scores = [85, 92]
# 默认 zip 会在最短处截断, zip_longest 用 fillvalue 填充
zipped = zip_longest(names, scores, fillvalue="缺考")
print("zip_longest 结果:", list(zipped))
# [('Alice', 85), ('Bob', 92), ('Charlie', '缺考')]

7. takewhile / dropwhile — 条件筛选
--------------------------------------

from itertools import takewhile, dropwhile

nums = [2, 4, 6, 7, 8, 9, 10]

# takewhile: 拿取满足条件的元素, 遇到第一个不满足就停止
print("takewhile 偶数:", list(takewhile(lambda x: x % 2 == 0, nums)))
# [2, 4, 6]

# dropwhile: 丢弃满足条件的元素, 遇到第一个不满足就保留后续全部
print("dropwhile 偶数:", list(dropwhile(lambda x: x % 2 == 0, nums)))
# [7, 8, 9, 10]

8. pairwise (Python 3.10+) / batched (Python 3.12+)
------------------------------------------------------

from itertools import pairwise, batched

# pairwise: 返回连续重叠对 (s -> (s0,s1), (s1,s2), ...)
items = [1, 2, 3, 4, 5]
print("pairwise 相邻对:", list(pairwise(items)))
# [(1,2), (2,3), (3,4), (4,5)]
# 应用: 检查序列是否单调递增
print("是否单调递增:", all(a < b for a, b in pairwise(items))) # True

# batched: 将迭代器分成指定大小的块
data_batches = list(batched(range(10), 3))
print("batched 分块:", data_batches)
# [(0,1,2), (3,4,5), (6,7,8), (9,)]

9. 自定义迭代器链式模式
-------------------------

def custom_iterator_chain(data):
"""链式组合多个 itertools 操作"""
from itertools import filterfalse, compress, starmap
# filterfalse: 保留不满足条件的元素
return (
data
# 这里用函数组合展示模式, 实际需顺序调用
)

# 典型流水线模式
def pipeline_example(nums):
"""数据处理流水线: 过滤 -> 变换 -> 分组 -> 聚合"""
# 过滤正数
positives = filter(lambda x: x > 0, nums)
# 映射平方
squares = map(lambda x: x ** 2, positives)
# 按奇偶分组
sorted_squares = sorted(squares, key=lambda x: x % 2)
grouped = groupby(sorted_squares, key=lambda x: x % 2)
result = {}
for is_even, group in grouped:
result["偶数" if is_even == 0 else "奇数"] = list(group)
return result

print("流水线结果:", pipeline_example([-3, -2, 1, 2, 3, 4]))

# compress: 按选择器筛选
data = ["a", "b", "c", "d"]
selectors = [1, 0, 1, 0]
print("compress 筛选:", list(compress(data, selectors))) # ['a', 'c']

# starmap: 用 * 解包参数
pairs = [(2, 3), (4, 5), (6, 7)]
print("starmap 求和:", list(starmap(lambda a, b: a + b, pairs)))
# [5, 9, 13]

总结: itertools 函数组合可构建声明式数据管道, 避免中间列表, 代码更简洁高效.