进阶篇 / 第 11 章

第 11 章 · 推导式、Lambda 与高阶函数

推导式：用一行表达循环

推导式（comprehension）是 Python 最有辨识度的语法之一，也是最容易「炫技过度」的地方。先看清它的本质：它就是一个表达式，所以可以直接赋值、可以当参数、可以嵌套。

三种基本形态

comprehension_basics.py · 四种形态与等价循环

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

# ---------- 列表推导式 ----------
# 循环写法
squares_loop = []
for x in nums:
    squares_loop.append(x * x)

# 推导式写法（快得多：没有反复调用 append）
squares = [x * x for x in nums]
print(squares)          # [1, 4, 9, 16, 25, 36]
print(squares_loop)     # 同样的结果

# 带条件筛选
evens = [x for x in nums if x % 2 == 0]
print(evens)            # [2, 4, 6]

# 带条件表达式（三元）
labels = ["偶" if x % 2 == 0 else "奇" for x in nums]
print(labels)           # ['奇', '偶', '奇', '偶', '奇', '偶']

# 多重 for（等价于嵌套循环，先外后内）
pairs = [(i, j) for i in range(3) for j in range(i, 3)]
print(pairs)            # [(0, 0), (0, 1), (0, 2), (1, 1), (1, 2), (2, 2)]

# 双重筛选：两个 if 的顺序有意义
flat = [x * y for x in [1, 2, 3] for y in [10, 20] if x != 2 and y != 10]
print(flat)             # [20, 40, 60]


# ---------- 集合推导式：自动去重 ----------
dup = [1, 2, 2, 3, 3, 3]
uniq = {x for x in dup}
print(uniq)             # {1, 2, 3}
letters = {ch for ch in "hello world"}
print(sorted(letters))  # [' ', 'd', 'h', 'l', 'o', 'r', 'w']


# ---------- 字典推导式 ----------
d1 = {"a": 1, "b": 2}
d2 = {"a": 10, "c": 30}

merged = {k: d1[k] + d2.get(k, 0) for k in set(d1) | set(d2)}
print(merged)           # {'a': 11, 'b': 2, 'c': 30}

# 反转字典
rev = {v: k for k, v in d1.items()}
print(rev)              # {1: 'a', 2: 'b'}

# 计数（最常见的字典推导式用法之一）
text = "hello world hello python"
counts = {ch: text.count(ch) for ch in set(text)}
print(sorted(counts.items(), key=lambda kv: -kv[1])[:3])
# [('l', 4), ('o', 4), ('h', 3)] 附近


# ---------- 生成器表达式：惰性版 ----------
gen = (x * x for x in range(1_000_000))
print(type(gen))        # <class 'generator'>

性能对比实测

import time

data = list(range(2_000_000))

t0 = time.perf_counter()
a = [x * 2 for x in data]          # 列表推导式
t1 = time.perf_counter()

b = []
for x in data:                      # 传统 for + append
    b.append(x * 2)
t2 = time.perf_counter()

c = list(map(lambda x: x * 2, data))   # map + lambda
t3 = time.perf_counter()

d = list(x * 2 for x in data)       # 生成器表达式物化
t4 = time.perf_counter()

print(f"列表推导式 : {t1 - t0:.4f}s")
print(f"for+append : {t2 - t1:.4f}s")
print(f"map+lambda : {t3 - t2:.4f}s")
print(f"生成器物化 : {t4 - t3:.4f}s")
# 典型结论：列表推导式 ≈ map+lambda > for+append > 生成器物化
# 且列表推导式峰值内存远低于 map/生成器物化（中间不产生额外列表）

推导式的进阶用法

从多个序列并行遍历

names = ["张三", "李四", "王五"]
scores = [88, 92, 79]
ages = [20, 25, 30]

# zip + 推导式
table = [(n, s, a) for n, s, a in zip(names, scores, ages)]
print(table)
# [('张三', 88, 20), ('李四', 92, 25), ('王五', 79, 30)]

# 批量格式化成一行文本
lines = [f"{n:<4s} {s:>3d}分 {a}岁" for n, s, a in zip(names, scores, ages)]
for line in lines:
    print(line)

# 字典形式
grade = {n: ("A" if s >= 90 else "B" if s >= 80 else "C")
         for n, s in zip(names, scores)}
print(grade)   # {'张三': 'B', '李四': 'A', '王五': 'C'}

嵌套数据结构的展平

data = [
    {"name": "苹果", "tags": ["水果", "甜"]},
    {"name": "汽车", "tags": ["工具", "大件"]},
    {"name": "钢琴", "tags": ["乐器"]},
]

# 从嵌套结构中抽取所有标签
all_tags = [t for item in data for t in item["tags"]]
print(all_tags)   # ['水果', '甜', '工具', '大件', '乐器']

# 去重
print(set(all_tags))   # {'水果', '甜', '工具', '大件', '乐器'}

# 按标签分组（字典推导式 + groupby 风格）
tag2name = {
    tag: [item["name"] for item in data if tag in item["tags"]]
    for tag in set(all_tags)
}
print(tag2name)
# {'水果': ['苹果'], '甜': ['苹果'], '工具': ['汽车'], '大件': ['汽车'], '乐器': ['钢琴']}

# 多层展平（深度不限的嵌套列表）
nested = [1, [2, [3, [4, [5]]]], 6]
def flatten(lst):
    out = []
    for item in lst:
        if isinstance(item, list):
            out.extend(flatten(item))
        else:
            out.append(item)
    return out
print(flatten(nested))    # [1, 2, 3, 4, 5, 6]

推导式 vs 生成器：选哪个

lambda 表达式

lambda 用于创建匿名的单表达式函数，语法极简：lambda 参数列表: 返回值表达式。

lambda_basics.py · lambda 等价于 def

# ---------- 普通函数 ----------
def add(a, b):
    return a + b


# ---------- lambda 表达式 ----------
add_lambda = lambda a, b: a + b

print(add(1, 2), add_lambda(1, 2))   # 3 3
print(add.__name__, add_lambda.__name__)   # add <lambda>  ← 元信息差别

# ---------- 带默认值（模拟可选参数） ----------
greet = lambda name, greeting="你好": f"{greeting}，{name}！"
print(greet("小明"))                 # 你好，小明！
print(greet("小明", greeting="早安")) # 早安，小明！

# ---------- 多参数、不限参数 ----------
add_many = lambda *args: sum(args)
print(add_many(1, 2, 3, 4))         # 10

# ---------- 结合内置排序函数（最常见的正当用途） ----------
people = [("王五", 79), ("张三", 88), ("李四", 92)]

# 按分数升序
by_score = sorted(people, key=lambda p: p[1])
print(by_score)      # [('王五', 79), ('张三', 88), ('李四', 92)]

# 多级排序：先按分数降序，分数相同按姓名字典序
people2 = [("bob", 90), ("alice", 90), ("carol", 85)]
srt = sorted(people2, key=lambda p: (-p[1], p[0]))
print(srt)           # [('alice', 90), ('bob', 90), ('carol', 85)]

lambda 的正确使用时机

社区对 lambda 的争论由来已久。Python 之父 Guido 曾称它「不值得的快捷方式」。争议的本质其实归结为一句话：lambda 应该只用在「定义后立即使用一次」的地方。

✅ 恰当用法

❌ 不恰当用法

# ❌ 反例一：需要复用时用 lambda（应该用 def）
square = lambda x: x * x
print(square(4), square(5))   # 能用，但丢了名字、无法被其他函数 import

# ✅ 正解
def square(x):
    return x * x

# ❌ 反例二：复杂逻辑塞进 lambda（可读性灾难）
result = lambda d: {k: v * 2 if isinstance(v, int) else [x for x in v if x] if isinstance(v, list) else v for k, v in d.items()}

# ✅ 正解：拆成有名字的函数
def normalize_value(v):
    if isinstance(v, int):
        return v * 2
    if isinstance(v, list):
        return [x for x in v if x]
    return v

def normalize(d):
    return {k: normalize_value(v) for k, v in d.items()}

# ❌ 反例三：lambda 有副作用（几乎总是设计错误）
nums = [1, 2, 3]
for_each = lambda: [nums.append(x * 10) for x in nums]

# ✅ 正解：用返回函数的方式（闭包）替代，语义清晰
def make_adder(n):
    def adder(x):
        return x + n
    return adder

add5 = make_adder(5)
print(add5(3))   # 8

map：高阶映射

map(function, iterable, ...) 把函数应用到每个元素上，返回一个惰性迭代器。

map_demo.py · map 的四种形式

# ---------- 基本用法 ----------
names = ["alice", "bob", "carol"]

# 传统循环
up = []
for n in names:
    up.append(n.upper())

# map + 方法引用（不需要 lambda！）
up2 = list(map(str.upper, names))
print(up, up2)         # 一样

# map + lambda
up3 = list(map(lambda n: n.upper(), names))
print(up3)

# ---------- 多个可迭代对象：并行逐元素应用 ----------
a = [1, 2, 3]
b = [10, 20, 30]
c = [100, 200, 300]

# 两个序列：函数收两个参数
add = map(lambda x, y: x + y, a, b)
print(list(add))      # [11, 22, 33]

# 三个序列
add3 = map(lambda x, y, z: x + y + z, a, b, c)
print(list(add3))     # [111, 222, 333]

# 长度不等时，以最短的为准
print(list(map(lambda x, y: x * y, [1, 2, 3, 4], [10, 20])))   # [10, 40]

# 常见技巧：多个序列求和
totals = list(map(lambda row: sum(row), zip(a, b, c)))
print(totals)         # [111, 222, 333]


# ---------- map 的返回值是迭代器（惰性） ----------
it = map(str, [1, 2, 3])
print(type(it))       # <class 'map'>
print(next(it))       # '1' —— 还没执行 str 的剩余部分

# 想重复使用必须物化
frozen = list(map(str, [1, 2, 3]))
print(frozen, frozen)   # ['1','2','3'] ['1','2','3']


# ---------- 实战：批量清洗数据 ----------
raw = ["  Alice ", "bob  ", " CAROL"]
cleaned = list(map(lambda s: s.strip().title(), raw))
print(cleaned)        # ['Alice', 'Bob', 'Carol']

# map 的链式：先转小写，再去空格
print(list(map(str.strip, map(str.lower, raw))))

filter：筛选

filter(predicate, iterable) 保留 predicate 返回真值的元素，同样返回迭代器。

nums = [1, -2, 3, -4, 5, 6, -7]

# ---------- 基本用法 ----------
positives = list(filter(lambda x: x > 0, nums))
print(positives)     # [1, 3, 5, 6]

# 用内置函数替代 lambda（更快）
nonzero = list(filter(None, nums))         # None 作为谓词 = 用真值判断
print(nonzero)       # [1, -2, 3, -4, 5, 6]

# 注意 filter(None) 判断的是「假值」，0 / "" / [] / None 都被剔除
mixed = [0, 1, "", "a", [], [1], None, False, True]
print(list(filter(None, mixed)))    # [1, 'a', [1], True]


# ---------- 多条件 ----------
def is_valid(item):
    """多条件判断写成命名函数，比 lambda 清晰。"""
    name, age = item
    return name and age >= 18


users = [("张三", 20), ("", 30), ("李四", 15), ("王五", 18)]
print(list(filter(is_valid, users)))
# [('张三', 20), ('王五', 18)]


# ---------- 对比三种写法 ----------
data = range(20)
even = [x for x in data if x % 2 == 0]                  # 列表推导式（推荐）
even2 = list(filter(lambda x: x % 2 == 0, data))        # filter
even3 = list(x for x in data if x % 2 == 0)            # 生成器表达式物化
print(even == even2 == even3)   # True


# ---------- filter + map 链式：先清洗后筛选 ----------
raw = ["1", "2", "x", "3", "y", "4"]
result = list(map(int, filter(str.isdigit, raw)))
print(result)    # [1, 2, 3, 4]

reduce：归约

reduce(function, iterable, initializer) 把序列逐步折叠成一个值。它在标准库里的位置有点尴尬——Python 3.x 中它被放进了 functools，但在 functools 文档里又被推荐用 functools.reduce（同名）而非内置的同名函数。

from functools import reduce

# ---------- 基本用法 ----------
nums = [1, 2, 3, 4, 5]

# 求和
print(reduce(lambda a, b: a + b, nums))        # 15
# 等价内置：sum(nums)

# 带初始值
print(reduce(lambda a, b: a + b, nums, 100))   # 115

# 求乘积（无内置替代）
import math
print(reduce(lambda a, b: a * b, nums))        # 120
# 或 math.prod(nums)  （Python 3.8+）

# 求最大值
print(reduce(lambda a, b: a if a > b else b, nums))   # 5
# 或 functools.reduce(max, nums)，或内置 max(nums)

# ---------- 组合多个序列 ----------
matrix = [(1, 2), (3, 4), (5, 6)]
s = reduce(lambda acc, (x, y): acc + x + y, matrix, 0)   # 解包语法在 lambda 参数里不可用，需括号

from functools import reduce
matrix = [(1, 2), (3, 4), (5, 6)]

# 用命名函数
def add_pair(acc, pair):
    return acc + pair[0] + pair[1]

print(reduce(add_pair, matrix, 0))   # 21

# 或 map + sum 更直白
print(sum(map(sum, matrix)))         # 21

# ---------- reduce 的实战场景：计算嵌套数据 ----------
from collections import defaultdict

orders = [
    {"user": "alice", "items": [{"price": 10, "qty": 2}, {"price": 5, "qty": 1}]},
    {"user": "bob",   "items": [{"price": 20, "qty": 1}]},
]


def order_total(order):
    """把一条订单的所有行项目金额折叠成总额。"""
    return reduce(lambda acc, item: acc + item["price"] * item["qty"],
                  order["items"], 0)


totals = {o["user"]: order_total(o) for o in orders}
print(totals)    # {'alice': 25, 'bob': 20}

# 但通常 sum 更直白
totals2 = {
    o["user"]: sum(i["price"] * i["qty"] for i in o["items"])
    for o in orders
}
print(totals2 == totals)   # True


# ---------- reduce 构建字典 ----------
pairs = [("a", 1), ("b", 2), ("a", 3)]

# 用 defaultdict + reduce
merged = defaultdict(int)
reduce(lambda acc, kv: acc.update({kv[0]: acc[kv[0]] + kv[1]}) or acc, pairs, merged)
print(dict(merged))     # {'a': 4, 'b': 2}

# 或 Counter（更简单）
from collections import Counter
c = Counter()
c.update(dict(pairs))
c.update(dict(pairs))
print(dict(c))         # {'a': 4, 'b': 2}

reduce 该用还是不该用

sorted 与排序技巧

sorted() 是最常用的高阶函数。它的 key 参数接收一个「从元素提取排序依据」的函数，reverse 控制方向。它总是返回新列表，不修改原数据。

# ---------- key 提取单一属性 ----------
words = ["banana", "apple", "cherry", "date"]
print(sorted(words, key=len))                    # 按长度：['date', 'apple', 'banana', 'cherry']
print(sorted(words))                             # 字典序
print(sorted(words, reverse=True))               # 降序

# ---------- 字典按键/按值排序 ----------
prices = {"苹果": 5.5, "香蕉": 3.0, "车厘子": 68.0}

# 按值升序
print(sorted(prices.items(), key=lambda kv: kv[1]))
# [('香蕉', 3.0), ('苹果', 5.5), ('车厘子', 68.0)]

# 按值降序
print(sorted(prices.items(), key=lambda kv: -kv[1]))
# [('车厘子', 68.0), ('苹果', 5.5), ('香蕉', 3.0)]

# 只要键
print(sorted(prices, key=prices.get, reverse=True))
# ['车厘子', '苹果', '香蕉']

# 只取值
print(sorted(prices.values(), reverse=True))    # [68.0, 5.5, 3.0]


# ---------- 多级排序：key 返回元组 ----------
records = [
    ("alice", 90, "math"),
    ("bob", 85, "physics"),
    ("carol", 90, "physics"),
    ("dave", 70, "math"),
]
# 先按分数降序，同分按姓名升序
print(sorted(records, key=lambda r: (-r[1], r[0])))
# [('alice', 90, 'math'), ('carol', 90, 'physics'), ('bob', 85, 'physics'), ('dave', 70, 'math')]


# ---------- 三个常见排序模式总结 ----------
# 1) 取元组第 k 位
pairs = [(3, "c"), (1, "a"), (2, "b")]
print(sorted(pairs, key=lambda p: p[0]))        # 按第 0 位
print(sorted(pairs, key=lambda p: p[1]))        # 按第 1 位

# 2) 多对象归一化为可比形式
items = [{"name": "a", "stock": 3}, {"name": "b", "stock": 1}]
print(sorted(items, key=lambda d: d["stock"]))

# 3) operator.itemgetter：比 lambda 更快，可指定多列
from operator import itemgetter, attrgetter
print(sorted(pairs, key=itemgetter(0)))
print(sorted(pairs, key=itemgetter(1)))

# 对对象按属性排序用 attrgetter
class Emp:
    def __init__(self, name, salary):
        self.name, self.salary = name, salary
    def __repr__(self):
        return f"Emp({self.name},{self.salary})"

emps = [Emp("bob", 5000), Emp("alice", 8000), Emp("carol", 6000)]
print(sorted(emps, key=attrgetter("salary"), reverse=True))
# [Emp(alice,8000), Emp(carol,6000), Emp(bob,5000)]


# ---------- 其他排序参数 ----------
from dataclasses import dataclass


@dataclass
class Task:
    title: str
    priority: int
    done: bool = False


tasks = [Task("写文档", 2), Task("修 bug", 1), Task("回邮件", 3)]
# done 的排后面，其余按 priority 升序
tasks.sort(key=lambda t: (t.done, t.priority))
print([t.title for t in tasks])   # ['修 bug', '写文档', '回邮件']
# 注意：list.sort 是原地排序，返回 None；sorted 是返回新列表

functools 模块

functools 提供了大量「操作函数」的工具函数，是高阶编程的瑞士军刀。

partial：偏函数

from functools import partial
import functools

# ---------- 绑定位置参数 ----------
def greet(greeting, name, punct="!"):
    return f"{greeting}，{name}{punct}"


hello = partial(greet, "你好")
print(hello("小明"))                     # 你好，小明!
print(hello("小红", punct="。"))         # 你好，小红。

# partial 会把参数按位置从左往右填充，绑定关键字参数同理
print(functools.partial(greet, name="固定")(greeting="嗨"))
# 嗨，固定!


# ---------- 实战：max/min 配合 partial ----------
# max 支持 key，这里用 partial 预先绑定一个属性提取器
class Ticket:
    def __init__(self, id, price):
        self.id, self.price = id, price

tickets = [Ticket(1, 50), Ticket(2, 30), Ticket(3, 80)]
cheapest = min(tickets, key=lambda t: t.price)
expensive = max(tickets, key=lambda t: t.price)
print(cheapest.id, expensive.id)   # 2 3


# ---------- 实战：排序统一 key 函数 ----------
people = [
    {"name": "alice", "age": 30},
    {"name": "bob", "age": 25},
    {"name": "carol", "age": 35},
]
by_age = lambda p: p["age"]                 # 本质是 partial(attrgetter, "age")
by_name = lambda p: p["name"]

print(sorted(people, key=by_age))
print(sorted(people, key=by_name, reverse=True))


# ---------- 实战：实现固定参数的可调用对象 ----------
class Repeat:
    """把字符串重复 n 次的函数工厂（partial 的手写等价物）。"""
    def __init__(self, n):
        self.n = n
    def __call__(self, text):
        return text * self.n


r = Repeat(3)
print(r("abc"))   # abcabcabc

缓存装饰器对比

import functools


# 1) cache：无上限，3.9+ 推荐
@functools.cache
def fib_cached(n: int) -> int:
    return n if n < 2 else fib_cached(n - 1) + fib_cached(n - 2)

# 2) lru_cache(maxsize=128)：有上限，按最近最少淘汰（LRU）
@functools.lru_cache(maxsize=128)
def fib_lru(n: int) -> int:
    return n if n < 2 else fib_lru(n - 1) + fib_lru(n - 2)

# 3) maxsize=None 时 lru_cache 等价于 cache
@functools.lru_cache(maxsize=None)
def fib_unbounded(n: int) -> int:
    return n if n < 2 else fib_unbounded(n - 1) + fib_unbounded(n - 2)


print(fib_cached(60))     # 1548008755920，瞬时
print(fib_lru.cache_info())     # CacheInfo(...)
print(fib_unbounded.cache_clear())   # 清空缓存


# 带自定义 key：只有列表参数时用元组包裹
@functools.lru_cache(maxsize=None)
def sum_of_list(nums: tuple) -> int:
    return sum(nums)

print(sum_of_list((1, 2, 3)))
print(sum_of_list.cache_info())

total_ordering 与 cmp_to_key

import functools


# 1) total_ordering：只实现 __eq__ 和 __lt__，自动补齐其他比较
@functools.total_ordering
class Version:
    def __init__(self, v: str):
        self.parts = tuple(int(x) for x in v.split("."))

    def __eq__(self, other):
        return self.parts == other.parts

    def __lt__(self, other):
        return self.parts < other.parts

    def __repr__(self):
        return ".".join(map(str, self.parts))


v1, v2 = Version("1.2.3"), Version("1.10.0")
print(v1 < v2, v1 <= v2, v1 != v2, v2 > v1)   # True True True True
# 若未用 total_ordering，v2 > v1 会报 TypeError


# 2) cmp_to_key：用「比较函数」替代 key 函数
#    经典场景：多字段自定义比较（如按多列排序）
@functools.cmp_to_key
def compare_people(a, b):
    """返回负数/0/正数，表示 a 排 b 之前/相等/之后。"""
    if a[1] != b[1]:          # 先比分数
        return b[1] - a[1]    # 分数高的在前（降序）
    return a[0].compare(b[0]) # 姓名升序


people = [("bob", 85), ("alice", 90), ("carol", 85)]
print(sorted(people, key=compare_people))
# [('alice', 90), ('carol', 85), ('bob', 85)]

operator 模块：函数化的运算符

operator 模块把常见的操作封装成函数对象，从而可以直接传给 map、reduce、sorted，避免写 lambda。

import operator

# ---------- itemgetter / attrgetter ----------
# itemgetter 可以取多个键，返回元组 → 天然实现多级排序
rows = [
    {"name": "alice", "dept": "tech", "salary": 9000},
    {"name": "bob", "dept": "tech", "salary": 7000},
    {"name": "carol", "dept": "sales", "salary": 8000},
]
# 先按部门排序，再按薪资排序（itemgetter 返回元组）
print(sorted(rows, key=operator.itemgetter("dept", "salary")))
# tech-bob(7000) → tech-alice(9000) → sales-carol
# 用 itemgetter 比 lambda -r["salary"], r["dept"] 更快

# 单独取一列（把字典列表转成值列表）
names = list(map(operator.itemgetter("name"), rows))
print(names)   # ['alice', 'bob', 'carol']


# ---------- itemgetter 取下标 ----------
matrix = [[1, 2, 3], [4, 5, 6]]
print(list(map(operator.itemgetter(2), matrix)))   # [3, 6]


# ---------- methodcaller：调用对象上的方法 ----------
words = ["alice", "bob"]
print(list(map(operator.methodcaller("upper"), words)))   # ['ALICE', 'BOB']
print(list(map(operator.methodcaller("count", "a"), words)))  # ['a' 的个数]


# ---------- 用 operator 替代 lambda 做归约 ----------
from functools import reduce
nums = [3, 1, 4, 1, 5]
print(reduce(operator.add, nums, 0))     # 14
print(reduce(operator.mul, nums, 1))     # 60
print(reduce(operator.and_, [True, True, False]))   # False（= all）
print(reduce(operator.or_, [False, False, True]))   # True  （= any）

itertools：惰性迭代器工厂

itertools 提供了大量组合与操作可迭代对象的高阶函数，全部返回惰性迭代器，非常适合处理大数据或无限序列。

无限序列

import itertools

# 计数、循环、重复：无限，靠 islice 取有限段
counter = itertools.count(10, 2)         # 10, 12, 14, ...
print(list(itertools.islice(counter, 5)))    # [10, 12, 14, 16, 18]

cycler = itertools.cycle("ABC")              # 无限循环
print(list(itertools.islice(cycler, 7)))     # ['A','B','C','A','B','C','A']

repeat = itertools.repeat("x", 3)             # 重复 3 次（count=3）
print(list(repeat))                          # ['x', 'x', 'x']
print(list(itertools.islice(itertools.repeat("y"), 4)))   # ['y','y','y','y']

组合函数：组合与排列

import itertools

letters = ["A", "B", "C"]

# ---------- combinations：组合（不考虑顺序） ----------
print(list(itertools.combinations(letters, 2)))
# [('A','B'), ('A','C'), ('B','C')]

# combinations：r=3 取 2 组，每组不重复
print(list(itertools.combinations([1, 2, 3, 4], 2)))
# 6 组二元组合

# ---------- permutations：排列（考虑顺序） ----------
print(list(itertools.permutations(letters, 2)))
# [('A','B'), ('A','C'), ('B','A'), ('B','C'), ('C','A'), ('C','B')]

# ---------- combinations_with_replacement：可重复组合 ----------
print(list(itertools.combinations_with_replacement([1, 2], 2)))
# [(1,1), (1,2), (2,2)]

# ---------- product：笛卡尔积 ----------
print(list(itertools.product([1, 2], "AB")))
# [(1,'A'), (1,'B'), (2,'A'), (2,'B')]

# repeat=n 参数模拟多序列笛卡尔积
print(list(itertools.product([0, 1], repeat=3)))   # 8 个三位二进制组合

# ---------- 真子集：全部非空子集 ----------
print(list(itertools.combinations([1, 2, 3], 2)))
# 所有二元子集；若要全部非空子集（含自身与全集），常用下面这种
def all_subsets(items):
    return itertools.chain.from_iterable(
        itertools.combinations(items, r) for r in range(1, len(items) + 1)
    )

print(list(all_subsets([1, 2, 3])))
# [(1,), (2,), (3,), (1,2), (1,3), (2,3), (1,2,3)]

筛选类：懒加载的过滤

import itertools

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

# takewhile：取到第一个不满足为止（一旦 False 就停）
print(list(itertools.takewhile(lambda x: x < 6, data)))     # [1,2,3,4,5]
print(list(itertools.takewhile(lambda x: x % 3 != 0, data)))  # [1,2]

# dropwhile：丢弃开头满足条件的（一旦 False 就继续）
print(list(itertools.dropwhile(lambda x: x < 6, data)))     # [6,7,8,9,10]

# filterfalse：取不满足条件的
print(list(itertools.filterfalse(lambda x: x % 2, data)))    # 偶数 [2,4,6,8,10]

# compress：按选择器筛选（选择器为真才保留数据）
selector = [1, 0, 1, 1, 0, 1, 1, 0, 1, 1]
print(list(itertools.compress(data, selector)))
# 取 selector 为真的位置的 data：1,3,4,6,7,9,10


# ---------- 累积类：前缀聚合 ----------
print(list(itertools.accumulate([1, 2, 3, 4])))          # [1,3,6,10]（前缀和）
print(list(itertools.accumulate([1, 2, 3, 4], max)))       # [1,2,3,4]（前缀最大）

running = itertools.accumulate([("a", 3), ("b", 1), ("c", 4)])
for name, cum in running:
    print(name, cum)      # a 3 / b 3 / c 4


# ---------- 分组：groupby ----------
# 注意：groupby 要求数据已按 key 排好序（相邻的相同 key 会被分到一组）
def show_groups(key, items):
    """演示 groupby 的正确用法：先排序，再分组。"""
    for k, g in itertools.groupby(sorted(items, key=key), key=key):
        print(k, list(g))

show_groups(lambda s: s[0], ["apple", "avocado", "banana", "blueberry"])
# a ['apple', 'avocado'] / b ['banana', 'blueberry']


# 去重（保留相邻不同，去掉连续重复）
print(list(itertools.groupby("aabbcda")))      # 需要配合压缩
compressed = [k for k, _ in itertools.groupby("aabbcda")]
print(compressed)                              # ['a','b','c','d']

groupby 的关键前提

from collections import defaultdict

# 全局分组的正确做法一：defaultdict
scores = {"alice": 90, "bob": 85, "carol": 90, "dave": 88}
by_score = defaultdict(list)
for name, s in scores.items():
    by_score[s].append(name)

# 同分排序输出
for s in sorted(by_score, reverse=True):
    print(s, sorted(by_score[s]))
# 90 ['alice', 'carol'] / 88 ['dave'] / 85 ['bob']

# 正确做法二：sorted + groupby
import itertools
items = sorted(scores.items(), key=lambda kv: -kv[0])
for s, group in itertools.groupby(items, key=lambda kv: kv[0]):
    print(s, [n for n, _ in group])

其他实用工具

import itertools

# zip_longest：长度不等时用默认值填充
a = [1, 2, 3]
b = ["a", "b"]
print(list(itertools.zip_longest(a, b, fillvalue="?")))
# [(1,'a'), (2,'b'), (3,'?')]

# chain：把多个可迭代对象串起来（惰性）
print(list(itertools.chain([1, 2], "ab", (3, 4))))   # [1,2,'a','b',3,4]

# chain.from_iterable：展平一层
nested = [[1, 2], [3, 4], [5, 6]]
print(list(itertools.chain.from_iterable(nested)))     # [1,2,3,4,5,6]
# 等价于内置 sum（数值场景）
print(sum(nested))   # 21

# islice：切片（惰性）
data = range(100)
print(list(itertools.islice(data, 10, 20)))       # 10..19
print(list(itertools.islice(data, 0, 10, 2)))     # 0,2,4,6,8

# pairwise：相邻元素两两配对
print(list(itertools.pairwise([1, 2, 3, 4, 5])))  # [(1,2),(2,3),(3,4),(4,5)]

# batched：分批
print(list(itertools.batched(range(7), 3)))      # [(0,1,2),(3,4,5),(6,)]


# ---------- 实战：按长度分词统计 ----------
text = "the quick brown fox jumps over the lazy dog the fox"
words = text.split()
by_len = defaultdict(list)
for w in words:
    by_len[len(w)].append(w)
for ln in sorted(by_len):
    print(f"长度 {ln}: {sorted(set(by_len[ln]))}")
# 长度 3: ['fox', 'the'] / 长度 5: ['quick'] / 长度 4: ['jump', 'lazy']

高阶函数实战：数据处理管线

下面把本章所有工具串成一个真实的数据清洗管线，展示它们如何组合。

pipeline.py · 组合 map/filter/sorted/operator 的完整管线

from collections import Counter, defaultdict
from dataclasses import dataclass
from functools import reduce
from itertools import groupby
import operator


@dataclass(frozen=True)
class Sale:
    """一条销售记录。"""
    region: str
    category: str
    amount: float
    qty: int

    @property
    def revenue(self) -> float:
        return self.amount * self.qty


RAW = [
    Sale("华东", "电子", 120.0, 3),
    Sale("华北", "服饰", 80.0, 5),
    Sale("华东", "食品", 25.5, 20),
    Sale("华南", "电子", 300.0, 1),
    Sale("华北", "电子", 150.0, 2),
    Sale("华东", "服饰", 60.0, 8),
    Sale("华南", "食品", 12.0, 100),
    Sale("华北", "食品", 18.0, 60),
]

# ---------- 第 1 步：筛选（filter + 命名谓词） ----------
def is_significant(sale: Sale) -> bool:
    """只保留营收超过 200 的记录。"""
    return sale.revenue >= 200


significant = list(filter(is_significant, RAW))
print(f"有效记录: {len(significant)} 条")
# 输出有效记录: 5 条

# ---------- 第 2 步：变换（map） ----------
enriched = list(map(lambda s: (s.region, s.category, s.revenue), significant))

# ---------- 第 3 步：分组聚合（defaultdict + reduce 或 sum） ----------
revenue_by_region = defaultdict(float)
for region, _cat, rev in enriched:
    revenue_by_region[region] += rev

print("各区域营收:")
for region, rev in sorted(revenue_by_region.items(), key=operator.itemgetter(1), reverse=True):
    print(f"  {region}: {rev:.1f}")
# 华北 300.0 / 华东 360.0 / 华南 300.0


# ---------- 第 4 步：找出 Top 品类（sorted + itemgetter） ----------
category_totals = defaultdict(float)
for _r, cat, rev in enriched:
    category_totals[cat] += rev

top2 = sorted(category_totals.items(), key=lambda kv: -kv[1])[:2]
print("Top 品类:", [c for c, _ in top2])    # ['电子', '服饰']


# ---------- 第 5 步：combine-style 统计 ----------
def tally_by_attr(attr: str):
    """返回一个按指定属性计数的高阶函数（函数返回函数）。"""
    key = operator.attrgetter(attr)

    def tally(sales):
        return Counter(key(s) for s in sales)
    return tally


count_by_region = tally_by_attr("region")
print("按区域计数:", dict(count_by_region(significant)))
# {'华东': 2, '华北': 2, '华南': 1}

count_by_category = tally_by_attr("category")
print("按品类计数:", dict(count_by_category(significant)))


# ---------- 第 6 步：groupby 分组统计 ----------
sorted_by_region = sorted(significant, key=operator.attrgetter("region"))
summary = {}
for region, group in groupby(sorted_by_region, key=operator.attrgetter("region")):
    group = list(group)
    summary[region] = {
        "count": len(group),
        "total": reduce(operator.add, [g.revenue for g in group], 0.0),
        "avg": round(sum(g.revenue for g in group) / len(group), 1),
    }

print("区域汇总:")
for region in sorted(summary):
    info = summary[region]
    print(f"  {region}: {info['count']} 笔, 合计 {info['total']:.1f}, 均值 {info['avg']}")


# ---------- 第 7 步：函数组合器（reduce 版 pipeline） ----------
PIPELINE_STEPS = [
    filter(is_significant),                       # 可迭代 -> 可迭代
    lambda xs: map(lambda s: s, xs),              # 恒等（演示）
    lambda xs: (s.revenue for s in xs),           # 提取营收
]

revenues = reduce(lambda f, g: g(f), PIPELINE_STEPS, RAW)
print("管线结果:", [round(r, 1) for r in revenues])
# [360.0, 400.0, 300.0, 300.0, 240.0]

函数式风格的边界

函数式编程不是银弹。理解它「什么时候好用、什么时候添乱」比会写更重要。

应该用函数式风格的场景

• 数据转换管线：一连串纯变换，每一步输入输出清晰——非常适合 map/filter/推导式组合。
• 不可变数据处理：数据转换不修改原对象（生成新对象），天然避免副作用。
• 批量操作多个集合：zip + map 处理「并行序列」最为简洁。
• 需要组合复用的小工具：partial、operator.itemgetter 让 key 提取逻辑可复用。
• 声明式排序/分组：sorted(key=...)、groupby 把意图直接写在代码里。

不适合函数式风格的场景

• 复杂状态变化：需要维护可变状态、多个变量相互影响时，函数式反而增加认知负担。
• 副作用密集：涉及大量 IO、数据库操作、UI 更新的逻辑，隐式数据流反而难追踪。
• 嵌套过深：三层以上的推导式嵌套或函数嵌套 lambda 会严重损害可读性。
• 调试困难时：没有函数名、没有断点友好结构的匿名代码，出错时定位成本高。
• 需要早退出时：reduce 无法提前终止，而普通循环可以 break。

函数式 vs 面向过程：同一任务的两种写法

# 场景：找出所有有效用户的邮箱域名分布

users = [
    {"name": "alice", "email": "alice@example.com", "active": True},
    {"name": "bob", "email": "bob@test.org", "active": False},
    {"name": "carol", "email": "carol@example.com", "active": True},
    {"name": "dave", "email": "dave@company.cn", "active": True},
]


# ---------- 函数式风格 ----------
from collections import Counter

domains = Counter(
    u["email"].split("@")[1]
    for u in users
    if u["active"]
)
print("函数式:", dict(domains))
# {'example.com': 2, 'company.cn': 1}


# ---------- 面向过程风格 ----------
from collections import defaultdict
proc_domains = defaultdict(int)
for u in users:
    if not u["active"]:
        continue
    domain = u["email"].split("@")[1]
    proc_domains[domain] += 1
print("过程式:", dict(proc_domains))
# 同上结果

# 两者都很好。选择依据：团队习惯 + 任务复杂度

最佳实践速查

十大 Pythonic 写法

import itertools
import operator
from collections import Counter, defaultdict
from functools import reduce, partial

data = [1, 2, 3, 4, 5, 6, 7, 8]
people = [("alice", 30), ("bob", 25), ("carol", 35)]

# 1. 用 sum/map 代替手动循环累加
print(sum(map(operator.mul, data, data)))          # 204  Σx²

# 2. 用 any/all 表达「存在/全部」
print(any(x > 7 for x in data), all(x > 0 for x in data))   # True True

# 3. 用 enumerate 同时拿下标与值
for i, v in enumerate(data[:3]):
    print(i, v)

# 4. 用 zip 打包多个序列
for (name, age), score in zip(people, [90, 80, 85]):
    print(name, age, score)

# 5. 用 Counter 一行统计频次
print(Counter("hello world").most_common(2))       # [('l', 4), ('o', 4)]

# 6. 用 defaultdict 免去初始化判断
d = defaultdict(list)
for name, age in people:
    d[age // 10].append(name)
print(dict(d))

# 7. 用 itemgetter 做多级排序
print(sorted(people, key=operator.itemgetter(1), reverse=True))

# 8. 用 itertools.groupby 分组（记得先排序）
for age, g in itertools.groupby(sorted(people, key=operator.itemgetter(1)),
                                key=operator.itemgetter(1)):
    print(age, [n for n, _ in g])

# 9. 用 partial 复用固定参数
print(list(map(partial(operator.add, 100), data)))   # 101..108

# 10. 用生成器表达式 + sum 表达「过滤后聚合」，比双重循环更快更清晰
print(sum(x for x in data if x % 2))               # 20

代码审查自查表

• □ 遍历+变换+筛选的场景用推导式，而不是 map+filter。
• □ 求和/最值不用 reduce，优先内置函数。
• □ lambda 只在 key 函数、单次回调处使用；复杂逻辑一律 def。
• □ 多级排序用 itemgetter 或 key 返回元组，而不是嵌套 lambda。
• □ groupby 前先 sorted。
• □ 生成器表达式用于大数据一次性遍历；需要复用时用列表。
• □ 推导式嵌套不超过两层；更深就用循环 + 辅助函数。
• □ 所有函数式代码都通过「三个月后我还能看懂吗」的自检。

本章小结

• 推导式有四种：列表 []、集合 {}、字典 {k:v}、生成器 ()；都是表达式，可嵌套、可赋值；列表推导式最快最常用。
• 性能：列表推导式 > map+lambda > for+append > 生成器物化；但生成器内存占用最低。
• lambda 是匿名单表达式函数；正当用途只有「sorting key」「简单 map/filter 回调」「GUI/配置钩子」；需要复用或复杂逻辑时用 def。
• map 返回惰性 map 对象，可配多个可迭代对象；通常可用列表推导式替代，map 的独有场景是有现成函数名、多序列并行、保持惰性。
• filter 第一个参数必须是函数（不是条件），filter(None, xs) 做真值筛选。
• reduce 做归约，已移入 functools；求和/最值/乘积请优先用 sum/max/math.prod，reduce 只用于自定义归约。
• sorted 返回新列表、list.sort 原地；key 返回元组即可多级排序，降序用取负值。
• functools：reduce、cache/lru_cache、partial、cmp_to_key、total_ordering、wraps。
• operator：itemgetter/attrgetter 做多级排序、methodcaller、运算符函数（add、mul）配 reduce。
• itertools：count/cycle/repeat（无限）、combinations/permutations/product、takewhile/dropwhile/filterfalse/compress、accumulate、groupby（须先 sorted）、chain/islice/batched/pairwise/zip_longest。
• 边界：函数式在数据转换管线、并行序列、排序分组场景最合适；涉及复杂状态、副作用、深层嵌套时应回到直白的循环。
• 终极原则：可读性 > 简洁性 > 函数式纯度。能写清楚的命令式代码就是好代码。

练习题

1. 推导式四连：对列表 nums = [1, -2, 3, 0, -5, 8, 12, -7, 4]，分别用列表/集合/字典/生成器四种推导式求出：(a) 所有正数的平方；(b) 去重后的绝对值；(c) {数字: 是否偶数}；(d) 偶数的和（惰性计算）。

   参考答案

   nums = [1, -2, 3, 0, -5, 8, 12, -7, 4]
   squares = [x * x for x in nums if x > 0]        # [1, 9, 64, 144, 16]
   uniq_abs = {abs(x) for x in nums if x != 0}     # {1,2,3,4,5,7,8,12}
   parity = {x: x % 2 == 0 for x in nums}          # {1: False, -2: True, 3: False, 0: True, ...}
   even_sum = sum(x for x in nums if x % 2 == 0)    # 24（-2+0+8+12+4）
   

   要点：字典推导式里用表达式作值，实现「元素 → 派生属性」的映射；生成器表达式不需要方括号，直接交给 sum 消费。

2. 多级排序实战：给定学生列表 [{"name":"张三","class":"A","score":88}, {"name":"李四","class":"B","score":92}, {"name":"王五","class":"A","score":95}, {"name":"赵六","class":"B","score":88}]，要求先按班级升序，同班按分数降序。用 sorted + key 实现，并输出形如 A: 王五(95), 张三(88) | B: 李四(92), 赵六(88) 的结果。

   参考答案要点sorted(students, key=lambda s: (s["class"], -s["score"]))；或用 itemgetter("class") 配合额外排序。格式化输出用 itertools.groupby 按班级分组后 join。注意 reverse=True 会同时反转班级顺序，所以必须用取负值只反转分数。

3. 用 itertools 重写：把下面用嵌套循环实现的「生成所有长度 3 的密码（字符集 A-Z）」改为 itertools 写法，并计算总数。

   result = []
   for a in "ABCDEFGHIJKLMNOPQRSTUVWXYZ":
       for b in "ABCDEFGHIJKLMNOPQRSTUVWXYZ":
           for c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ":
               if a != b and b != c and a != c:   # 三个字符互不相同
                   result.append(a + b + c)
   

   参考答案

   from itertools import permutations
   letters = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
   result = ["".join(p) for p in permutations(letters, 3)]
   print(len(result))     # 26*25*24 = 15600
   

   要点：permutations 天然保证元素不重复。若要允许重复则用 product(letters, repeat=3)（26³ = 17576）。product 对应原来的三层循环（含相等情况）。

4. groupby 正确性：给定 records = [("b", "x"), ("a", "y"), ("b", "z"), ("a", "w")]，(a) 直接用 groupby 按首字母分组，观察结果；(b) 修正它使分组正确；(c) 用 defaultdict 再实现一次。

   参考答案(a) 直接分组得到 b:[('b','x')]、a:[('a','y')]、b:[('b','z')]、a:[('a','w')] —— b 和 a 各被拆成两组，错误。(b) 先 sorted(records, key=lambda r: r[0]) 再 groupby，得到 b 有 2 条、a 有 2 条。(c) d=defaultdict(list); for k,v in records: d[k].append(v)，无需排序。全要点：groupby 只聚合相邻相同 key。

5. 高阶函数工厂：写一个函数 make_multiplier(factor) 返回一个把数字乘以 factor 的函数；再写 make_pipeline(*steps)，返回一个能把数据依次通过所有步骤的函数。用它们组合出「先 +10 再 ×3 再取绝对值」的管线。

   参考答案要点

   def make_multiplier(factor):
       def mul(x): return x * factor
       return mul
   
   def make_pipeline(*steps):
       def run(data):
           for s in steps:
               data = s(data)
           return data
       return run
   
   import operator
   pipe = make_pipeline(lambda x: x + 10, make_multiplier(3), abs)
   print(pipe(-5))     # ( -5+10 )*3 = 15 → abs = 15
   print(pipe(-20))    # (-20+10)*3 = -30 → abs = 30
   

   要点：函数作为返回值是「高阶」的另一面；这个管线内部顺序是 run(abs(make_multiplier(3)(x+10)))，与数学的 |3(x+10)| 相同。

6. 词频分析器：给定一段中英混合文本，要求：(a) 统计最常见的 5 个词（忽略大小写和标点）；(b) 输出每个词首次出现的下标；(c) 把所有不重复单词按长度分组。请至少用两种标准库方案实现。

   参考答案要点(a) 方案一：Counter(w.lower() for w in re.findall(r"[A-Za-z]+", text)).most_common(5)；方案二：collections.Counter(...) 配合 str.translate 去标点。(b) 用 enumerate 遍历去重后的有序结果，或 {w: i for i, w in enumerate(words)}（字典保留首次插入位置）。(c) defaultdict(list) 或 itertools.groupby(sorted(set(words), key=len))，注意 groupby 前必须 sorted。

7. reduce 重构：把下面三个用了 reduce 但有更简洁替代的代码改成内置函数版本，并解释为什么替换更好。

   from functools import reduce
   import operator
   
   nums = [4, 1, 7, 3]
   total = reduce(operator.add, nums, 0)
   mx = reduce(max, nums)
   prod = reduce(operator.mul, nums, 1)
   joined = reduce(operator.add, ["a", "b", "c"], "")
   

   参考答案要点total = sum(nums)；mx = max(nums)；prod = math.prod(nums)（3.8+）；joined = "".join(["a","b","c"])。更好的原因：① 内置用 C 实现，速度快一个数量级；② 意图一目了然，reduce(operator.add, ...) 需要读者额外解码；③ 无需导入 functools/operator，减少依赖。

8. 缓存装饰器对比实验：写一个不带缓存的递归斐波那契 fib_slow(n) 与带 functools.cache 的 fib_fast(n)，分别调用 n=30 和 n=40，记录耗时与 cache_info()，解释命中/未命中的含义。

   参考答案要点fib_slow(35) 在纯 Python 下需要数秒到数十秒（调用次数呈指数增长 ≈ 2⁴⁰ 量级）；fib_fast(35) 瞬时完成。cache_info() 中 hits 是「参数相同直接复用结果」的次数，misses 是「需实际计算」的次数；缓存把 O(2ⁿ) 的调用数压到 O(n)。还可对比 @cache（无上限）与 @lru_cache(maxsize=128)（有上限会淘汰旧项）。

9. 数据清洗管线：给定 raw = [" Alice 30 ", "bob,25", "CAROL 41", " ", "dave 35x"]，用本章所学（推导式/filter/map/regex 辅助）清洗为 [("Alice", 30), ("Bob", 25), ("Carol", 41), ("Dave", 35)]，并保证非法项被丢弃而不报错。

   参考答案要点先用 str.strip 去掉空白、re.split(r"[ ,]+", s) 拆成两段、re.search(r"[0-9]+", ...) 提数字；再用 filter 过滤长度不足的项、用 map 转 title() 和 int。关键是先整体 strip 再判空，否则全空格的项会漏网；可以写一个 parse_record(s) -> tuple | None 命名函数返回 None 表示失败，再用 filter(lambda r: r is not None, map(parse_record, raw))——这里 map 与 filter 的组合是合理的，因为 parse_record 已有名字。

10. 综合重构：下面这段代码「能跑但很糟」。请重写为高质量 Python，要求：不使用 reduce 替代 sum；不用嵌套 3 层以上的结构；所有转换逻辑有名字；处理空输入；返回类型清晰。

    import json
    
    def report(data_json):
        d = json.loads(data_json)
        res = {}
        for k in d["items"] if "items" in d else []:
            if k["qty"] > 0 and k["price"] >= 0:
                if k["category"] not in res:
                    res[k["category"]] = 0
                res[k["category"]] += k["qty"] * k["price"]
        return {c: round(res[c], 2) for c in sorted(res.keys()) if res[c] > 100}
    

    参考答案要点拆成三个命名函数：load_items(data_json) -> list[dict]（负责 json 解析与取 items，解析失败返回 []）、valid_items(items) -> Iterator[dict]（生成器过滤 qty>0 and price>=0，利用第 10 章的惰性）、totals_by_category(items) -> dict[str, float]（用 defaultdict(float) 累加）。最后 filter_totals(totals, threshold=100) 用推导式输出。类型标注为 def report(data_json: str) -> dict[str, float]。