itertools 模块：构建高效迭代器的工具箱

import itertools

# ========== 基础用法：生成自增序号 ==========
counter = itertools.count(1)          # 从 1 开始，步长为 1
print([next(counter) for _ in range(5)])  # [1, 2, 3, 4, 5]

# ========== 自定义步长 ==========
even = itertools.count(0, 2)          # 从 0 开始，步长为 2
print([next(even) for _ in range(5)])     # [0, 2, 4, 6, 8]

# ========== 实际场景：为日志行添加行号 ==========
logs = ["ERROR: connection refused", "WARN: retrying", "INFO: connected"]
for i, msg in zip(itertools.count(1), logs):
    print(f"[{i}] {msg}")
# 输出:
# [1] ERROR: connection refused
# [2] WARN: retrying
# [3] INFO: connected

# ========== 浮点步长（谨慎使用，可能累积误差） ==========
float_counter = itertools.count(0, 0.1)
print([round(next(float_counter), 1) for _ in range(5)])  # [0, 0.1, 0.2, 0.3, 0.4]

import itertools

# ========== 基础用法 ==========
colors = ['red', 'green', 'blue']
cycler = itertools.cycle(colors)
print([next(cycler) for _ in range(7)])
# ['red', 'green', 'blue', 'red', 'green', 'blue', 'red']

# ========== 实际场景：轮询分配任务到多个 Worker ==========
workers = ['worker-A', 'worker-B', 'worker-C']
tasks = ['task-1', 'task-2', 'task-3', 'task-4', 'task-5']

assignments = list(zip(itertools.cycle(workers), tasks))
print(assignments)
# [('worker-A', 'task-1'), ('worker-B', 'task-2'), ('worker-C', 'task-3'),
#  ('worker-A', 'task-4'), ('worker-B', 'task-5')]

# ========== 注意：永远不会停止，必须外部截断 ==========
# 下面这行会无限循环——不要直接运行！
# for x in itertools.cycle([1, 2, 3]):
#     print(x)

import itertools

# ========== 有限重复 ==========
print(list(itertools.repeat('hello', 3)))  # ['hello', 'hello', 'hello']

# ========== 实际场景：为 map() 提供常量参数 ==========
# 将多个数字乘以同一个系数 2.5
prices = [10, 20, 30]
with_tax = list(map(lambda p, rate: p * rate, prices, itertools.repeat(1.13)))
print(with_tax)  # [11.3, 22.6, 33.9]

# ========== 等价于生成器表达式（但 repeat 更清晰地表达意图） ==========
with_tax2 = [p * 1.13 for p in prices]

import itertools
import operator

# ========== 默认：累加 ==========
nums = [1, 2, 3, 4, 5]
print(list(itertools.accumulate(nums)))
# [1, 3, 6, 10, 15]

# ========== 累乘 ==========
print(list(itertools.accumulate(nums, operator.mul)))
# [1, 2, 6, 24, 120]

# ========== 使用 initial 参数（Python 3.8+） ==========
print(list(itertools.accumulate(nums, initial=100)))
# [100, 101, 103, 106, 110, 115]

# ========== 实际场景：计算累计收益率 ==========
daily_returns = [0.01, -0.02, 0.03, 0.01, -0.005]
cumulative = itertools.accumulate(daily_returns, lambda total, r: round(total * (1 + r), 4))
print(list(cumulative))
# [0.01, 0.0098, 0.0101, 0.0102, 0.0101]

import itertools

# ========== chain：串联多个序列 ==========
a = [1, 2, 3]
b = ['a', 'b', 'c']
c = (True, False)
print(list(itertools.chain(a, b, c)))
# [1, 2, 3, 'a', 'b', 'c', True, False]

# ========== chain.from_iterable：扁平化一层嵌套 ==========
nested = [[1, 2], [3, 4], [5, 6]]
print(list(itertools.chain.from_iterable(nested)))
# [1, 2, 3, 4, 5, 6]

# ========== 实际场景：合并多个数据源的结果 ==========
db_results = [("Alice", 25), ("Bob", 30)]
api_results = [("Charlie", 28)]
cache_results = [("Dave", 35)]
all_users = list(itertools.chain(db_results, api_results, cache_results))
print(all_users)
# [('Alice', 25), ('Bob', 30), ('Charlie', 28), ('Dave', 35)]

import itertools

# ========== 基础用法 ==========
data = ['A', 'B', 'C', 'D', 'E']
selectors = [True, False, 1, 0, True]  # 1 视为真，0 视为假
print(list(itertools.compress(data, selectors)))
# ['A', 'C', 'E']

# ========== 实际场景：按布尔掩码提取字段 ==========
columns = ['id', 'name', 'email', 'deleted_at', 'created_at']
visibility = [True, True, True, False, False]  # 隐藏 deleted_at 和 created_at
visible_columns = list(itertools.compress(columns, visibility))
print(visible_columns)  # ['id', 'name', 'email']

import itertools

data = [1, 3, 5, 2, 4, 6, 7]

# ========== dropwhile：跳过开头符合条件的元素 ==========
print(list(itertools.dropwhile(lambda x: x < 5, data)))
# [5, 2, 4, 6, 7] — 跳过 1,3（都<5），遇到 5 后全部保留

# ========== takewhile：取开头符合条件的元素 ==========
print(list(itertools.takewhile(lambda x: x < 5, data)))
# [1, 3] — 1<5 保留, 3<5 保留, 遇到 5 停止

# ========== 实际场景：解析日志文件，跳过开头注释行 ==========
log_lines = [
    "# ===== LOG START =====",
    "# Generated: 2026-06-05",
    "INFO: process started",
    "ERROR: timeout",
    "WARN: retry attempt 1",
]
# 跳过所有以 # 开头的注释行
content_lines = itertools.dropwhile(lambda line: line.startswith("#"), log_lines)
print(list(content_lines))
# ['INFO: process started', 'ERROR: timeout', 'WARN: retry attempt 1']

import itertools

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

# ========== 过滤掉所有零（保留非零） ==========
print(list(itertools.filterfalse(lambda x: x == 0, nums)))
# [1, 2, 3, 4, 5]

# ========== 等效于 filter + not ==========
print(list(filter(lambda x: x != 0, nums)))  # [1, 2, 3, 4, 5]

# ========== 实际场景：分离有效数据和无效数据 ==========
records = [100, None, 200, None, 300, None]
valid = list(itertools.filterfalse(lambda x: x is None, records))
print(valid)  # [100, 200, 300]

import itertools

# ========== 基础用法：按首字母分组 ==========
words = ['apple', 'ant', 'bear', 'bat', 'cat', 'car']
words.sort(key=lambda w: w[0])  # groupby 要求预排序！
groups = itertools.groupby(words, key=lambda w: w[0])
for letter, items in groups:
    print(f"{letter}: {list(items)}")
# 输出:
# a: ['apple', 'ant']
# b: ['bear', 'bat']
# c: ['cat', 'car']

# ========== 实际场景：按日期分组日志 ==========
logs = [
    ('2026-06-01', 'INFO', 'start'),
    ('2026-06-01', 'ERROR', 'timeout'),
    ('2026-06-02', 'INFO', 'restart'),
    ('2026-06-02', 'WARN', 'slow query'),
    ('2026-06-02', 'INFO', 'success'),
]
logs.sort(key=lambda x: x[0])  # 按日期排序
grouped_logs = itertools.groupby(logs, key=lambda x: x[0])
for date, entries in grouped_logs:
    entries_list = list(entries)
    print(f"{date}: {len(entries_list)} entries → {[e[1] for e in entries_list]}")
# 2026-06-01: 2 entries → ['INFO', 'ERROR']
# 2026-06-02: 3 entries → ['INFO', 'WARN', 'INFO']

import itertools

data = list(range(10))  # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

# ========== 等价于 data[2:7] ==========
print(list(itertools.islice(data, 2, 7)))  # [2, 3, 4, 5, 6]

# ========== 等价于 data[:5] ==========
print(list(itertools.islice(data, 5)))     # [0, 1, 2, 3, 4]

# ========== 等价于 data[2:9:3] ==========
print(list(itertools.islice(data, 2, 9, 3)))  # [2, 5, 8]

# ========== 与无限迭代器配合：取 count(10) 的前 5 个 ==========
print(list(itertools.islice(itertools.count(10), 5)))  # [10, 11, 12, 13, 14]

# ========== islice 不支持负数索引（不同于普通切片） ==========
# itertools.islice(data, -3, None)  # ❌ ValueError

import itertools

# ========== 基础用法 ==========
data = [1, 2, 3, 4, 5]
print(list(itertools.pairwise(data)))
# [(1, 2), (2, 3), (3, 4), (4, 5)]

# ========== 实际场景：计算相邻元素的差值 ==========
values = [10, 15, 22, 30, 25]
diffs = [b - a for a, b in itertools.pairwise(values)]
print(diffs)  # [5, 7, 8, -5]

import itertools

# ========== 基础用法 ==========
# 将每组 (a, b) 传入 pow(a, b)
pairs = [(2, 3), (3, 2), (4, 2)]
print(list(itertools.starmap(pow, pairs)))
# [8, 9, 16]  # 等价于 pow(2,3), pow(3,2), pow(4,2)

# ========== 与 map 的区别 ==========
def add(a, b):
    return a + b

# map: 需要手动解包每个元组
print(list(map(lambda p: add(p[0], p[1]), pairs)))
# starmap: 自动解包
print(list(itertools.starmap(add, pairs)))
# [5, 5, 6]

import itertools

source = [1, 2, 3]
it1, it2, it3 = itertools.tee(source, 3)

print(list(it1))  # [1, 2, 3]
print(list(it2))  # [1, 2, 3]
print(list(it3))  # [1, 2, 3]

# ========== 注意：tee 可能缓存大量数据 ==========
# 如果迭代器 A 和 B 的消费速度差异很大，
# 快的那一个会触发大量中间结果被缓存在内存中

import itertools

a = [1, 2, 3]
b = ['a', 'b']

# ========== 内置 zip：以最短为准，截断 ==========
print(list(zip(a, b)))  # [(1, 'a'), (2, 'b')]

# ========== zip_longest：以最长为标准，填充缺失值 ==========
print(list(itertools.zip_longest(a, b, fillvalue='缺')))
# [(1, 'a'), (2, 'b'), (3, '缺')]

# ========== 实际场景：对齐不同长度的数据列 ==========
names = ['Alice', 'Bob', 'Charlie']
scores = [95, 87]  # Charlie 缺考
for name, score in itertools.zip_longest(names, scores, fillvalue='缺考'):
    print(f"{name}: {score}")
# Alice: 95
# Bob: 87
# Charlie: 缺考

import itertools

# ========== 两个集合的笛卡尔积 ==========
colors = ['red', 'blue']
sizes = ['S', 'M', 'L']
print(list(itertools.product(colors, sizes)))
# [('red', 'S'), ('red', 'M'), ('red', 'L'),
#  ('blue', 'S'), ('blue', 'M'), ('blue', 'L')]

# ========== repeat 参数：单个集合的多次笛卡尔积 ==========
print(list(itertools.product('AB', repeat=2)))
# [('A', 'A'), ('A', 'B'), ('B', 'A'), ('B', 'B')]

# ========== 实际场景：生成测试用例的所有参数组合 ==========
browsers = ['Chrome', 'Firefox', 'Safari']
resolutions = ['1920x1080', '1366x768']
languages = ['zh-CN', 'en-US']

test_cases = list(itertools.product(browsers, resolutions, languages))
print(f"共生成 {len(test_cases)} 个测试用例")  # 12 个
for case in test_cases[:3]:
    print(f"  测试: {case}")
# 测试: ('Chrome', '1920x1080', 'zh-CN')
# 测试: ('Chrome', '1920x1080', 'en-US')
# 测试: ('Chrome', '1366x768', 'zh-CN')

import itertools

items = ['A', 'B', 'C']

# ========== 全排列 ==========
print(list(itertools.permutations(items)))
# [('A', 'B', 'C'), ('A', 'C', 'B'), ('B', 'A', 'C'),
#  ('B', 'C', 'A'), ('C', 'A', 'B'), ('C', 'B', 'A')]

# ========== 取 2 个元素的排列 ==========
print(list(itertools.permutations(items, 2)))
# [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]

# ========== 注意：元素值相同也视为不同的项 ==========
duplicates = ['A', 'A', 'B']
print(list(itertools.permutations(duplicates, 2)))
# [('A', 'A'), ('A', 'B'), ('A', 'A'), ('A', 'B'), ('B', 'A'), ('B', 'A')]

import itertools

items = ['A', 'B', 'C']

# ========== 取 2 个元素的组合 ==========
print(list(itertools.combinations(items, 2)))
# [('A', 'B'), ('A', 'C'), ('B', 'C')]

# ========== 与 permutations 的对比 ==========
# permutations(items, 2) 有 6 个结果（(A,B) 和 (B,A) 都算）
# combinations(items, 2) 有 3 个结果（(A,B) 和 (B,A) 只算一个）

# ========== 实际场景：从队伍中选出双打配对 ==========
players = ['选手A', '选手B', '选手C', '选手D']
pairings = list(itertools.combinations(players, 2))
print(pairings)
# [('选手A', '选手B'), ('选手A', '选手C'), ('选手A', '选手D'),
#  ('选手B', '选手C'), ('选手B', '选手D'), ('选手C', '选手D')]

import itertools

items = ['A', 'B', 'C']

# ========== 取 2 个元素的可重复组合 ==========
print(list(itertools.combinations_with_replacement(items, 2)))
# [('A', 'A'), ('A', 'B'), ('A', 'C'), ('B', 'B'), ('B', 'C'), ('C', 'C')]

# ========== 实际场景：投 2 个骰子（可重复的 6 种面值取 2 个） ==========
dice = list(itertools.combinations_with_replacement(range(1, 7), 2))
print(f"可能的点数组合数: {len(dice)}")  # 21 种

# ========== 三种组合函数的结果数量对比 ==========
n, r = 5, 3
print(f"permutations:    {len(list(itertools.permutations(range(n), r)))}")         # 60 = n!/(n-r)!
print(f"combinations:    {len(list(itertools.combinations(range(n), r)))}")          # 10 = n!/(r!(n-r)!)
print(f"combinations_wr: {len(list(itertools.combinations_with_replacement(range(n), r)))}")  # 35 = (n+r-1)!/(r!(n-1)!)

import itertools

def sliding_window(iterable, n: int):
    """生成大小为 n 的滑动窗口。"""
    # 创建 n 个独立的迭代器
    iterators = itertools.tee(iterable, n)
    # 每个迭代器依次向前偏移 0, 1, 2, ..., n-1 个位置
    staggered = (
        itertools.islice(it, i, None)
        for i, it in enumerate(iterators)
    )
    # 重新组合，生成 (it[0], it[1], ..., it[n-1])
    return zip(*staggered)

# ========== 使用示例：计算移动平均线 ==========
prices = [10, 12, 11, 14, 13, 15, 16, 14, 17]
window_size = 3

for window in sliding_window(prices, window_size):
    avg = sum(window) / window_size
    print(f"窗口 {window}: 均价 = {avg:.1f}")
# 窗口 (10, 12, 11): 均价 = 11.0
# 窗口 (12, 11, 14): 均价 = 12.3
# 窗口 (11, 14, 13): 均价 = 12.7
# ... 以此类推

# ========== Python 3.10+ 可用 pairwise 实现窗口大小为 2 的特例 ==========
for a, b in itertools.pairwise(prices):
    print(f"{a} → {b}: 变化 = {b - a:+d}")

import itertools
from statistics import mean

# 模拟数据：日期 + 销售额
transactions = [
    ('2026-06-01', 150),
    ('2026-06-01', 200),
    ('2026-06-01', 180),
    ('2026-06-02', 300),
    ('2026-06-02', 250),
    ('2026-06-03', 100),
    ('2026-06-03', 400),
    ('2026-06-03', 350),
    ('2026-06-03', 120),
]

# 按日期排序（groupby 的前提）
transactions.sort(key=lambda t: t[0])

# 分组聚合：计算每日销售总额和平均交易额
report = []
for date, records in itertools.groupby(transactions, key=lambda t: t[0]):
    amounts = [amt for _, amt in records]
    report.append({
        'date': date,
        'total': sum(amounts),
        'avg': mean(amounts),
        'count': len(amounts),
    })

for r in report:
    print(f"{r['date']}: 总额={r['total']}, 均额={r['avg']:.1f}, 笔数={r['count']}")
# 2026-06-01: 总额=530, 均额=176.7, 笔数=3
# 2026-06-02: 总额=550, 均额=275.0, 笔数=2
# 2026-06-03: 总额=970, 均额=242.5, 笔数=4

import itertools
from collections import defaultdict

# 数据：两种渠道的销售额
online_sales = [
    ('2026-06-01', 'online', 150),
    ('2026-06-02', 'online', 200),
    ('2026-06-02', 'online', 180),
    ('2026-06-03', 'online', 300),
]
store_sales = [
    ('2026-06-01', 'store', 400),
    ('2026-06-02', 'store', 350),
    ('2026-06-03', 'store', 500),
    ('2026-06-03', 'store', 250),
]

# 1. chain: 合并两个数据源
all_sales = itertools.chain(online_sales, store_sales)

# 2. 按日期排序（groupby 的前提）
sorted_sales = sorted(all_sales, key=lambda x: x[0])

# 3. groupby: 按日期分组
daily_summary = {}
for date, records in itertools.groupby(sorted_sales, key=lambda x: x[0]):
    # 将 group 迭代器转为列表以便多次遍历
    items = list(records)
    daily_summary[date] = {
        'total': sum(amt for _, _, amt in items),
        'online': sum(amt for _, ch, amt in items if ch == 'online'),
        'store': sum(amt for _, ch, amt in items if ch == 'store'),
    }

for date, summary in daily_summary.items():
    print(f"{date}: 总计={summary['total']}, 线上={summary['online']}, 门店={summary['store']}")
# 2026-06-01: 总计=550, 线上=150, 门店=400
# 2026-06-02: 总计=730, 线上=380, 门店=350
# 2026-06-03: 总计=1050, 线上=300, 门店=750

import itertools

# ❌ 错误：同一个迭代器不能遍历两次
data = [1, 2, 3, 4, 5]
evens = filterfalse(lambda x: x % 2 != 0, data)  # 过滤出偶数
result1 = list(evens)   # OK: [2, 4]
result2 = list(evens)   # 陷阱！返回 []，迭代器已耗尽
print(f"第一次: {result1}, 第二次: {result2}")

# ✅ 方案一：物化为列表再复用
evens_list = list(filterfalse(lambda x: x % 2 != 0, data))
print(list(evens_list))  # [2, 4]
print(list(evens_list))  # [2, 4]

# ✅ 方案二：用 tee 复制
evens = filterfalse(lambda x: x % 2 != 0, data)
it1, it2 = itertools.tee(evens)
print(list(it1))  # [2, 4]
print(list(it2))  # [2, 4]

import itertools

# ❌ 错误：未排序的 groupby
unsorted = ['apple', 'bear', 'ant', 'bat', 'cat', 'car']
groups = itertools.groupby(unsorted, key=lambda w: w[0])
for letter, items in groups:
    print(f"{letter}: {list(items)}")
# 输出:
# a: ['apple']          ← 两个 a 被分开了！
# b: ['bear']
# a: ['ant']            ← 这是第二个 a 组
# b: ['bat']
# c: ['cat', 'car']

# ✅ 正确：先排序
sorted_words = sorted(unsorted, key=lambda w: w[0])
groups = itertools.groupby(sorted_words, key=lambda w: w[0])
for letter, items in groups:
    print(f"{letter}: {list(items)}")
# a: ['ant', 'apple']   ← 所有 a 合并在一个组
# b: ['bat', 'bear']
# c: ['car', 'cat']

import itertools

# 场景：从日志流中提取错误信息，去重后取前 10 条
log_stream = iter([
    "INFO: ok", "ERROR: timeout", "ERROR: timeout",
    "WARN: slow", "INFO: ok", "ERROR: crash",
    "ERROR: disk full", "INFO: ok",
])

# 管道构建（每一步都是惰性的）
# 1. 只保留 ERROR
errors = filterfalse(lambda line: not line.startswith("ERROR:"), log_stream)
# 2. 去掉前缀
messages = map(lambda line: line.removeprefix("ERROR: "), errors)
# 3. 去重
seen = set()
unique_messages = filterfalse(lambda msg: msg in seen or seen.add(msg), messages)
# 4. 取前 10 条
top_errors = list(itertools.islice(unique_messages, 10))

print(top_errors)  # ['timeout', 'crash', 'disk full']

import itertools

# ❌ 过度使用 itertools 导致难以理解
result = list(itertools.compress(
    itertools.chain.from_iterable(
        map(lambda x: [x, x*2], range(10))
    ),
    itertools.cycle([True, False])
))

# ✅ 拆分为有意义的步骤，每个变量名说明意图
doubled = itertools.chain.from_iterable(
    (x, x * 2) for x in range(10)
)
every_other = itertools.compress(doubled, itertools.cycle([True, False]))
result = list(every_other)