高级技巧

一、高级 Python 学什么

掌握变量、数据结构、函数和类以后，下一阶段主要学习：

更简洁地处理数据
更合理地组织程序
更高效地处理大量任务
更安全地管理资源
更准确地表达类型
更方便地测试和维护

高级 Python 的核心不是把代码写得复杂，而是：

用更合适的语言特性，把复杂问题写得更简单。

主要内容包括：

解包和参数传递
推导式和生成器
迭代器
装饰器
闭包和高阶函数
上下文管理器
特殊方法
数据类
模式匹配
高级类型提示
异常组
并发编程
异步编程
缓存
性能分析
日志
测试
项目结构

二、序列解包

Python 可以把列表、元组等序列中的元素直接分配给多个变量。

stock = ("600519", "贵州茅台", 1500.0)

code, name, price = stock

print(code)
print(name)
print(price)

变量数量必须与元素数量一致。

1. 星号解包

使用 * 接收多个元素：

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

first, *middle, last = numbers

print(first)
print(middle)
print(last)

结果：

1
[2, 3, 4]
5

注意，星号变量得到的是列表。

2. 忽略不需要的值

stock = (
    "600519",
    "贵州茅台",
    1500.0,
    "白酒",
)

code, name, *_ = stock

通常使用 _ 表示该值不会使用。

只取首尾：

first, *_, last = numbers

3. 嵌套解包

stock = (
    "600519",
    "贵州茅台",
    (1500.0, 1520.0),
)

code, name, (open_price, close_price) = stock

三、合并数据结构

1. 合并列表

list_a = [1, 2]
list_b = [3, 4]

result = [*list_a, *list_b, 5]

结果：

[1, 2, 3, 4, 5]

2. 合并字典

base = {
    "code": "600519",
    "name": "贵州茅台",
}

price_info = {
    "price": 1500.0,
    "change": 1.5,
}

stock = {
    **base,
    **price_info,
}

也可以使用 Python 3.9 以上支持的 |：

stock = base | price_info

如果键重复，后面的值覆盖前面的值：

a = {"price": 100}
b = {"price": 120}

result = a | b

结果：

{"price": 120}

原地更新：

a |= b

四、推导式高级用法

1. 列表推导式

numbers = [
    number ** 2
    for number in range(1, 11)
]

带条件：

even_squares = [
    number ** 2
    for number in range(1, 11)
    if number % 2 == 0
]

2. 条件表达式

labels = [
    "偶数" if number % 2 == 0 else "奇数"
    for number in range(1, 6)
]

注意两种条件位置的区别。

过滤：

[
    number
    for number in numbers
    if number > 0
]

二选一转换：

[
    number if number > 0 else 0
    for number in numbers
]

3. 嵌套推导式

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

展开二维列表：

flattened = [
    number
    for row in matrix
    for number in row
]

结果：

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

复杂推导式可读性较差时，应改成普通循环。

4. 字典推导式

stocks = [
    {
        "code": "600519",
        "name": "贵州茅台",
    },
    {
        "code": "000001",
        "name": "平安银行",
    },
]

stock_map = {
    stock["code"]: stock["name"]
    for stock in stocks
}

结果：

{
    "600519": "贵州茅台",
    "000001": "平安银行",
}

5. 集合推导式

prefixes = {
    code[:2]
    for code in [
        "600519",
        "603259",
        "000001",
        "002594",
    ]
}

五、海象运算符

Python 3.8 以上支持 :=，称为赋值表达式或海象运算符。

普通写法：

text = input("请输入内容：")

while text != "quit":
    print(text)
    text = input("请输入内容：")

使用海象运算符：

while (
    text := input("请输入内容：")
) != "quit":
    print(text)

1. 减少重复计算

普通写法：

data = get_data()
length = len(data)

if length > 10:
    print(length)

简化：

if (
    length := len(get_data())
) > 10:
    print(length)

2. 推导式中使用

texts = [
    "Python",
    "",
    "Java",
    "  ",
    "Go",
]

clean_texts = [
    clean
    for text in texts
    if (clean := text.strip())
]

结果：

["Python", "Java", "Go"]

不要为了少写一行而滥用 :=。

六、高级排序

1. key 参数

stocks = [
    {
        "name": "贵州茅台",
        "price": 1500,
    },
    {
        "name": "平安银行",
        "price": 11,
    },
    {
        "name": "宁德时代",
        "price": 300,
    },
]

result = sorted(
    stocks,
    key=lambda stock: stock["price"],
)

2. 多字段排序

先按行业排序，再按价格降序：

stocks = [
    {
        "name": "A公司",
        "industry": "银行",
        "price": 10,
    },
    {
        "name": "B公司",
        "industry": "白酒",
        "price": 100,
    },
    {
        "name": "C公司",
        "industry": "银行",
        "price": 20,
    },
]

result = sorted(
    stocks,
    key=lambda stock: (
        stock["industry"],
        -stock["price"],
    ),
)

3. operator.itemgetter

from operator import itemgetter

按照字典字段排序：

result = sorted(
    stocks,
    key=itemgetter("price"),
)

多字段：

result = sorted(
    stocks,
    key=itemgetter(
        "industry",
        "price",
    ),
)

4. operator.attrgetter

对象列表：

from dataclasses import dataclass
from operator import attrgetter


@dataclass
class Stock:
    code: str
    price: float

排序：

stocks = [
    Stock("600519", 1500),
    Stock("000001", 11),
]

result = sorted(
    stocks,
    key=attrgetter("price"),
)

七、collections 高级容器

collections 提供了许多实用数据结构。

1. Counter 计数器

from collections import Counter

industries = [
    "银行",
    "白酒",
    "银行",
    "半导体",
    "银行",
]

counts = Counter(industries)

print(counts)

结果：

Counter({
    "银行": 3,
    "白酒": 1,
    "半导体": 1,
})

查看出现次数最多的元素：

print(counts.most_common(2))

结果：

[
    ("银行", 3),
    ("白酒", 1),
]

更新计数：

counts.update([
    "银行",
    "白酒",
])

2. defaultdict

普通字典分组：

groups = {}

for stock in stocks:
    industry = stock["industry"]

    if industry not in groups:
        groups[industry] = []

    groups[industry].append(stock)

使用 defaultdict：

from collections import defaultdict

groups: defaultdict[
    str,
    list[dict[str, object]],
] = defaultdict(list)

for stock in stocks:
    groups[stock["industry"]].append(stock)

统计：

counts = defaultdict(int)

for industry in industries:
    counts[industry] += 1

3. deque 双端队列

from collections import deque

queue = deque()

右侧添加：

queue.append("A")

左侧添加：

queue.appendleft("B")

右侧弹出：

queue.pop()

左侧弹出：

queue.popleft()

deque 从左侧删除元素的效率高于列表。

限制最大长度：

recent_prices = deque(
    maxlen=5
)

for price in [
    10,
    11,
    12,
    13,
    14,
    15,
]:
    recent_prices.append(price)

最终只保留最近五个值。

适合：

队列
滑动窗口
最近记录
广度优先搜索

4. ChainMap

将多个字典组合成一个逻辑视图：

from collections import ChainMap

default_config = {
    "timeout": 10,
    "retries": 3,
}

user_config = {
    "timeout": 30,
}

config = ChainMap(
    user_config,
    default_config,
)

print(config["timeout"])
print(config["retries"])

优先使用前面的字典。

八、迭代协议

能够被 for 遍历的对象称为可迭代对象。

例如：

list
tuple
str
dict
set
range

for 循环内部会先调用：

iter(object)

得到迭代器，再不断调用：

next(iterator)

1. 手动迭代

numbers = [10, 20, 30]

iterator = iter(numbers)

print(next(iterator))
print(next(iterator))
print(next(iterator))

再次调用：

next(iterator)

会抛出：

StopIteration

2. 自定义迭代器

class Countdown:
    def __init__(
        self,
        start: int,
    ) -> None:
        self.current = start

    def __iter__(self):
        return self

    def __next__(self) -> int:
        if self.current <= 0:
            raise StopIteration

        value = self.current
        self.current -= 1

        return value

使用：

for number in Countdown(5):
    print(number)

3. 可迭代对象和迭代器

可迭代对象：

可以生成迭代器

迭代器：

可以通过 next() 逐个返回数据

迭代器通常只能消费一次。

九、生成器

生成器是创建迭代器的简单方式。

def countdown(start: int):
    while start > 0:
        yield start
        start -= 1

使用：

for number in countdown(5):
    print(number)

yield 会：

返回一个值
暂停函数
保存当前状态
下次继续执行

1. 生成器表达式

列表推导式：

squares = [
    number ** 2
    for number in range(1_000_000)
]

生成器表达式：

squares = (
    number ** 2
    for number in range(1_000_000)
)

列表会一次性保存全部结果。

生成器按需产生数据，更节省内存。

2. yield from

普通写法：

def flatten(groups):
    for group in groups:
        for item in group:
            yield item

使用 yield from：

def flatten(groups):
    for group in groups:
        yield from group

调用：

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

print(list(flatten(groups)))

3. 分批处理数据

def batch[
    T
](
    items: list[T],
    size: int,
):
    if size <= 0:
        raise ValueError(
            "size 必须大于 0"
        )

    for index in range(
        0,
        len(items),
        size,
    ):
        yield items[
            index:index + size
        ]

使用：

codes = [
    "600519",
    "000001",
    "300750",
    "002594",
    "603259",
]

for group in batch(
    codes,
    size=2,
):
    print(group)

结果：

["600519", "000001"]
["300750", "002594"]
["603259"]

4. 读取大文件

不推荐：

content = file.read()

大文件可能占用大量内存。

推荐逐行处理：

def read_lines(
    file_path: str,
):
    with open(
        file_path,
        "r",
        encoding="utf-8",
    ) as file:
        for line in file:
            yield line.rstrip("\n")

十、itertools

itertools 提供高效的迭代工具。

1. chain

合并多个可迭代对象：

from itertools import chain

result = chain(
    [1, 2],
    [3, 4],
    [5, 6],
)

print(list(result))

2. islice

对迭代器进行切片：

from itertools import islice

numbers = (
    number
    for number in range(100)
)

result = islice(
    numbers,
    10,
    20,
)

print(list(result))

3. groupby

按照连续相同键分组。

使用前通常需要先排序：

from itertools import groupby
from operator import itemgetter

stocks = sorted(
    stocks,
    key=itemgetter("industry"),
)

for industry, group in groupby(
    stocks,
    key=itemgetter("industry"),
):
    print(
        industry,
        list(group),
    )

4. combinations

生成组合：

from itertools import combinations

codes = [
    "600519",
    "000001",
    "300750",
]

for pair in combinations(
    codes,
    2,
):
    print(pair)

5. product

笛卡尔积：

from itertools import product

markets = ["上海", "深圳"]
directions = ["买入", "卖出"]

for item in product(
    markets,
    directions,
):
    print(item)

6. accumulate

累计计算：

from itertools import accumulate

changes = [
    10,
    -5,
    20,
    -3,
]

result = list(
    accumulate(changes)
)

print(result)

结果：

[10, 5, 25, 22]

十一、闭包

闭包是内部函数记住外层函数变量的机制。

def create_multiplier(
    factor: float,
):
    def multiply(
        value: float,
    ) -> float:
        return value * factor

    return multiply

使用：

double = create_multiplier(2)
triple = create_multiplier(3)

print(double(10))
print(triple(10))

闭包适合：

保存配置
创建函数工厂
装饰器
保存简单状态

1. nonlocal

def create_counter():
    count = 0

    def increase() -> int:
        nonlocal count
        count += 1
        return count

    return increase

使用：

counter = create_counter()

print(counter())
print(counter())
print(counter())

十二、高阶函数

接收函数作为参数，或者返回函数的函数，称为高阶函数。

def apply_operation(
    a: float,
    b: float,
    operation,
):
    return operation(a, b)

使用：

def add(
    a: float,
    b: float,
) -> float:
    return a + b


result = apply_operation(
    10,
    20,
    add,
)

1. functools.partial

固定函数的一部分参数：

from functools import partial


def calculate_tax(
    amount: float,
    rate: float,
) -> float:
    return amount * rate

创建固定税率函数：

calculate_vat = partial(
    calculate_tax,
    rate=0.13,
)

print(
    calculate_vat(1000)
)

2. reduce

from functools import reduce

numbers = [1, 2, 3, 4]

result = reduce(
    lambda a, b: a * b,
    numbers,
)

print(result)

结果：

很多场景中，普通循环比 reduce() 更清晰。

十三、装饰器

装饰器用于在不修改原函数代码的情况下增加功能。

1. 基础装饰器

from functools import wraps


def log_call(func):
    @wraps(func)
    def wrapper(
        *args,
        **kwargs,
    ):
        print(
            f"开始调用："
            f"{func.__name__}"
        )

        result = func(
            *args,
            **kwargs,
        )

        print(
            f"调用结束："
            f"{func.__name__}"
        )

        return result

    return wrapper

使用：

@log_call
def add(
    a: int,
    b: int,
) -> int:
    return a + b

2. 带参数的装饰器

from functools import wraps


def repeat(times: int):
    if times <= 0:
        raise ValueError(
            "times 必须大于 0"
        )

    def decorator(func):
        @wraps(func)
        def wrapper(
            *args,
            **kwargs,
        ):
            result = None

            for _ in range(times):
                result = func(
                    *args,
                    **kwargs,
                )

            return result

        return wrapper

    return decorator

使用：

@repeat(3)
def greet(name: str) -> None:
    print(f"你好，{name}")

3. 执行时间装饰器

from functools import wraps
from time import perf_counter


def measure_time(func):
    @wraps(func)
    def wrapper(
        *args,
        **kwargs,
    ):
        start = perf_counter()

        try:
            return func(
                *args,
                **kwargs,
            )
        finally:
            elapsed = (
                perf_counter()
                - start
            )

            print(
                f"{func.__name__} "
                f"耗时："
                f"{elapsed:.6f} 秒"
            )

    return wrapper

4. 类装饰器

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(
        self,
        *args,
        **kwargs,
    ):
        self.count += 1

        print(
            f"调用次数："
            f"{self.count}"
        )

        return self.func(
            *args,
            **kwargs,
        )

使用：

@CountCalls
def say_hello() -> None:
    print("Hello")

十四、上下文管理器

上下文管理器负责自动获取和释放资源。

最常见的例子：

with open(
    "data.txt",
    "r",
    encoding="utf-8",
) as file:
    content = file.read()

退出 with 后，文件自动关闭。

1. 自定义上下文管理器

class Timer:
    def __enter__(self):
        from time import perf_counter

        self._start = perf_counter()

        return self

    def __exit__(
        self,
        exc_type,
        exc_value,
        traceback,
    ):
        from time import perf_counter

        self.elapsed = (
            perf_counter()
            - self._start
        )

        print(
            f"耗时："
            f"{self.elapsed:.6f} 秒"
        )

        return False

使用：

with Timer():
    total = sum(
        range(1_000_000)
    )

2. contextmanager

from contextlib import contextmanager
from time import perf_counter


@contextmanager
def timer():
    start = perf_counter()

    try:
        yield
    finally:
        elapsed = (
            perf_counter()
            - start
        )

        print(
            f"耗时："
            f"{elapsed:.6f} 秒"
        )

使用：

with timer():
    total = sum(
        range(1_000_000)
    )

3. suppress

忽略指定异常：

from contextlib import suppress

with suppress(
    FileNotFoundError
):
    file_path.unlink()

等价于：

try:
    file_path.unlink()
except FileNotFoundError:
    pass

不要忽略不明确的异常。

4. ExitStack

动态管理多个上下文：

from contextlib import ExitStack

file_names = [
    "a.txt",
    "b.txt",
    "c.txt",
]

with ExitStack() as stack:
    files = [
        stack.enter_context(
            open(
                name,
                "r",
                encoding="utf-8",
            )
        )
        for name in file_names
    ]

    contents = [
        file.read()
        for file in files
    ]

十五、特殊方法

特殊方法让自定义对象支持 Python 内置语法。

1. repr

class Stock:
    def __init__(
        self,
        code: str,
        price: float,
    ) -> None:
        self.code = code
        self.price = price

    def __repr__(self) -> str:
        return (
            "Stock("
            f"code={self.code!r}, "
            f"price={self.price!r}"
            ")"
        )

2. 比较方法

class Stock:
    def __init__(
        self,
        code: str,
        price: float,
    ) -> None:
        self.code = code
        self.price = price

    def __lt__(
        self,
        other: "Stock",
    ) -> bool:
        return self.price < other.price

现在可以：

stocks.sort()

3. getitem

让对象支持索引：

class StockCollection:
    def __init__(
        self,
        stocks: list,
    ) -> None:
        self._stocks = stocks

    def __getitem__(
        self,
        index,
    ):
        return self._stocks[index]

使用：

collection[0]
collection[1:3]

4. call

让对象可以像函数一样调用：

class PriceFilter:
    def __init__(
        self,
        min_price: float,
    ) -> None:
        self.min_price = min_price

    def __call__(
        self,
        stock: dict[str, object],
    ) -> bool:
        return (
            float(stock["price"])
            >= self.min_price
        )

使用：

price_filter = PriceFilter(
    min_price=100,
)

result = [
    stock
    for stock in stocks
    if price_filter(stock)
]

十六、数据类高级用法

from dataclasses import dataclass

1. slots

@dataclass(slots=True)
class Stock:
    code: str
    name: str
    price: float

特点：

限制动态添加属性
减少内存占用
适合大量对象

2. frozen

@dataclass(
    frozen=True,
    slots=True,
)
class StockCode:
    value: str

创建后不能修改：

code.value = "000001"

会报错。

3. order

@dataclass(order=True)
class Stock:
    price: float
    code: str
    name: str

会自动生成比较方法。

字段顺序决定比较顺序。

如果只想按价格比较：

from dataclasses import (
    dataclass,
    field,
)


@dataclass(order=True)
class Stock:
    price: float
    code: str = field(
        compare=False
    )
    name: str = field(
        compare=False
    )

4. kw_only

@dataclass(kw_only=True)
class Order:
    code: str
    price: float
    quantity: int

创建对象时必须使用关键字：

order = Order(
    code="600519",
    price=1500,
    quantity=100,
)

十七、结构模式匹配

Python 3.10 以上支持 match。

1. 基础匹配

def handle_status(
    status: int,
) -> str:
    match status:
        case 200:
            return "成功"

        case 404:
            return "不存在"

        case 500:
            return "服务器错误"

        case _:
            return "未知状态"

2. 匹配多个值

match status:
    case 200 | 201 | 204:
        print("请求成功")

    case 400 | 404:
        print("客户端错误")

    case _:
        print("其他状态")

3. 匹配序列

command = [
    "buy",
    "600519",
    "100",
]

match command:
    case [
        "buy",
        code,
        quantity,
    ]:
        print(
            f"买入 {code} "
            f"{quantity} 股"
        )

    case [
        "sell",
        code,
        quantity,
    ]:
        print(
            f"卖出 {code} "
            f"{quantity} 股"
        )

    case _:
        print("无效命令")

4. 匹配字典

event = {
    "type": "price",
    "code": "600519",
    "price": 1500.0,
}

match event:
    case {
        "type": "price",
        "code": code,
        "price": price,
    }:
        print(
            code,
            price,
        )

    case {
        "type": "news",
        "title": title,
    }:
        print(title)

    case _:
        print("未知事件")

5. 守卫条件

match stock:
    case {
        "code": code,
        "change": change,
    } if change >= 9.5:
        print(
            f"{code} 接近涨停"
        )

    case {
        "code": code,
        "change": change,
    } if change <= -9.5:
        print(
            f"{code} 接近跌停"
        )

    case _:
        print("普通波动")

6. 匹配类对象

from dataclasses import dataclass


@dataclass
class BuyOrder:
    code: str
    quantity: int


@dataclass
class SellOrder:
    code: str
    quantity: int

def handle_order(order) -> None:
    match order:
        case BuyOrder(
            code=code,
            quantity=quantity,
        ):
            print(
                f"买入 {code} "
                f"{quantity} 股"
            )

        case SellOrder(
            code=code,
            quantity=quantity,
        ):
            print(
                f"卖出 {code} "
                f"{quantity} 股"
            )

十八、高级类型提示

类型提示可以提高大型项目的可维护性。

1. 类型别名

Python 3.12 支持：

type StockCode = str
type Price = float
type StockData = dict[
    str,
    object,
]

使用：

def normalize_code(
    code: StockCode,
) -> StockCode:
    return code.strip()

2. 泛型函数

Python 3.12 新语法：

def first[T](
    items: list[T],
) -> T | None:
    if not items:
        return None

    return items[0]

3. 泛型类

class Repository[T]:
    def __init__(self) -> None:
        self._items: list[T] = []

    def add(
        self,
        item: T,
    ) -> None:
        self._items.append(item)

    def all(self) -> list[T]:
        return self._items.copy()

使用：

stock_repository = (
    Repository[Stock]()
)

4. TypedDict

用于描述固定结构的字典：

from typing import TypedDict


class StockDict(TypedDict):
    code: str
    name: str
    price: float

函数：

def show_stock(
    stock: StockDict,
) -> None:
    print(
        stock["code"],
        stock["name"],
        stock["price"],
    )

可选字段：

from typing import NotRequired


class StockDict(TypedDict):
    code: str
    name: str
    price: float
    industry: NotRequired[str]

5. Literal

限制参数只能取指定值：

from typing import Literal

type Direction = Literal[
    "buy",
    "sell",
]


def create_order(
    code: str,
    direction: Direction,
) -> None:
    pass

6. Protocol

描述对象需要具备的方法，而不要求继承指定父类。

from typing import Protocol


class Executable(Protocol):
    def execute(self) -> None:
        ...

函数：

def run_task(
    task: Executable,
) -> None:
    task.execute()

任何拥有 execute() 方法的对象都可以传入。

7. Callable

描述函数类型：

from collections.abc import Callable


def apply_filter(
    stocks: list[Stock],
    condition: Callable[
        [Stock],
        bool,
    ],
) -> list[Stock]:
    return [
        stock
        for stock in stocks
        if condition(stock)
    ]

8. overload

为同一函数描述多个调用方式：

from typing import overload


@overload
def parse_value(
    value: str,
    target: type[int],
) -> int:
    ...


@overload
def parse_value(
    value: str,
    target: type[float],
) -> float:
    ...


def parse_value(
    value: str,
    target: type[int] | type[float],
) -> int | float:
    return target(value)

十九、自定义异常

不要让所有错误都使用 ValueError。

class StockError(Exception):
    """股票业务异常。"""


class InvalidStockCodeError(
    StockError
):
    """股票代码无效。"""


class InvalidPriceError(
    StockError
):
    """股票价格无效。"""

使用：

def validate_code(
    code: str,
) -> str:
    if (
        len(code) != 6
        or not code.isdigit()
    ):
        raise InvalidStockCodeError(
            f"无效股票代码：{code}"
        )

    return code

捕获：

try:
    validate_code("ABC")
except InvalidStockCodeError as error:
    print(error)

二十、异常链

在处理底层异常时，可以保留原始原因。

def parse_price(
    value: str,
) -> float:
    try:
        return float(value)
    except ValueError as error:
        raise InvalidPriceError(
            f"价格格式错误：{value}"
        ) from error

异常信息会显示完整调用链。

不希望保留底层异常：

raise InvalidPriceError(
    "价格无效"
) from None

二十一、异常组

Python 3.11 以上支持 ExceptionGroup 和 except*。

errors = [
    ValueError("价格错误"),
    TypeError("类型错误"),
]

raise ExceptionGroup(
    "数据处理失败",
    errors,
)

分别捕获：

try:
    raise ExceptionGroup(
        "多个错误",
        [
            ValueError("值错误"),
            TypeError("类型错误"),
        ],
    )
except* ValueError as error:
    print(
        "捕获值错误：",
        error,
    )
except* TypeError as error:
    print(
        "捕获类型错误：",
        error,
    )

异常组在并发任务中尤其有用。

二十二、缓存

1. lru_cache

from functools import lru_cache


@lru_cache(maxsize=128)
def calculate(
    number: int,
) -> int:
    print(
        f"实际计算：{number}"
    )

    return number ** 2

重复调用：

print(calculate(10))
print(calculate(10))

第二次直接使用缓存。

查看缓存信息：

print(
    calculate.cache_info()
)

清空缓存：

calculate.cache_clear()

参数必须是可哈希对象。

2. cache

无限缓存：

from functools import cache


@cache
def fibonacci(
    number: int,
) -> int:
    if number < 2:
        return number

    return (
        fibonacci(number - 1)
        + fibonacci(number - 2)
    )

3. cached_property

from functools import cached_property


class Report:
    def __init__(
        self,
        values: list[float],
    ) -> None:
        self.values = values

    @cached_property
    def average(self) -> float:
        print("执行平均值计算")

        return (
            sum(self.values)
            / len(self.values)
        )

第一次访问时计算，后续直接返回缓存结果：

report.average
report.average

如果底层数据变化，需要手动删除缓存：

del report.average

二十三、并发编程基础

并发主要有三种方式：

多线程
多进程
异步编程

选择原则：

网络请求、文件等待   → 多线程或异步
CPU 密集计算         → 多进程
大量并发网络任务     → asyncio

二十四、多线程

from concurrent.futures import (
    ThreadPoolExecutor,
)
from time import sleep


def fetch_data(
    code: str,
) -> str:
    sleep(1)
    return f"{code} 数据"

并发执行：

codes = [
    "600519",
    "000001",
    "300750",
]

with ThreadPoolExecutor(
    max_workers=3,
) as executor:
    results = list(
        executor.map(
            fetch_data,
            codes,
        )
    )

print(results)

顺序与输入顺序一致。

1. submit 和 as_completed

from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed,
)

with ThreadPoolExecutor(
    max_workers=3,
) as executor:
    future_map = {
        executor.submit(
            fetch_data,
            code,
        ): code
        for code in codes
    }

    for future in as_completed(
        future_map
    ):
        code = future_map[future]

        try:
            result = future.result()
        except Exception as error:
            print(
                code,
                error,
            )
        else:
            print(
                code,
                result,
            )

2. 线程安全

多个线程修改共享数据可能出现竞争条件。

from threading import Lock

lock = Lock()
total = 0


def increase() -> None:
    global total

    with lock:
        total += 1

更好的方式是尽量减少共享可变状态。

二十五、多进程

from concurrent.futures import (
    ProcessPoolExecutor,
)


def calculate_square(
    number: int,
) -> int:
    return number ** 2


def main() -> None:
    numbers = list(
        range(1_000_000)
    )

    with ProcessPoolExecutor() as executor:
        results = list(
            executor.map(
                calculate_square,
                numbers,
                chunksize=1000,
            )
        )


if __name__ == "__main__":
    main()

多进程需要注意：

启动进程有成本
数据需要序列化
小任务不一定更快
Windows 和 macOS 应保留主入口判断

二十六、异步编程

异步适合大量等待型任务。

import asyncio

定义异步函数：

async def fetch_data(
    code: str,
) -> str:
    await asyncio.sleep(1)

    return f"{code} 数据"

运行：

async def main() -> None:
    result = await fetch_data(
        "600519"
    )

    print(result)


asyncio.run(main())

1. 并发执行多个协程

async def main() -> None:
    codes = [
        "600519",
        "000001",
        "300750",
    ]

    results = await asyncio.gather(
        *[
            fetch_data(code)
            for code in codes
        ]
    )

    print(results)

如果顺序执行，需要约三秒。

并发执行通常只需要约一秒。

2. TaskGroup

Python 3.11 以上支持结构化并发：

async def main() -> None:
    tasks = []

    async with asyncio.TaskGroup() as group:
        for code in [
            "600519",
            "000001",
            "300750",
        ]:
            task = group.create_task(
                fetch_data(code)
            )

            tasks.append(task)

    results = [
        task.result()
        for task in tasks
    ]

    print(results)

如果其中一个任务失败，其他任务会被统一管理。

3. 超时控制

async def main() -> None:
    try:
        async with asyncio.timeout(2):
            result = await fetch_data(
                "600519"
            )

            print(result)

    except TimeoutError:
        print("任务超时")

4. 限制并发数量

async def limited_fetch(
    code: str,
    semaphore: asyncio.Semaphore,
) -> str:
    async with semaphore:
        return await fetch_data(code)

async def main() -> None:
    semaphore = asyncio.Semaphore(5)

    results = await asyncio.gather(
        *[
            limited_fetch(
                code,
                semaphore,
            )
            for code in codes
        ]
    )

避免一次发起过多请求。

二十七、同步和异步选择

同步：

def function():
    pass

异步：

async def function():
    pass

异步不是自动变快。

只有存在大量等待时才有明显优势，例如：

网络请求
数据库查询
文件或消息队列等待
WebSocket
大量 API 调用

纯计算任务应优先考虑多进程。

二十八、日志

不要在正式项目中到处使用 print()。

import logging

基础配置：

logging.basicConfig(
    level=logging.INFO,
    format=(
        "%(asctime)s "
        "%(levelname)s "
        "%(name)s "
        "%(message)s"
    ),
)

获取日志器：

logger = logging.getLogger(
    __name__
)

记录日志：

logger.debug("调试信息")
logger.info("程序开始")
logger.warning("数据不完整")
logger.error("操作失败")
logger.critical("严重错误")

1. 记录异常

try:
    result = 10 / 0
except ZeroDivisionError:
    logger.exception(
        "计算失败"
    )

logger.exception() 会自动记录异常堆栈。

2. 参数化日志

二十九、性能计时

1. perf_counter

from time import perf_counter

start = perf_counter()

result = sum(
    range(1_000_000)
)

elapsed = (
    perf_counter()
    - start
)

print(elapsed)

2. timeit

from timeit import timeit

elapsed = timeit(
    "sum(range(1000))",
    number=10_000,
)

print(elapsed)

比较两种写法：

list_time = timeit(
    "[x * x for x in range(1000)]",
    number=10_000,
)

loop_time = timeit(
    """
result = []
for x in range(1000):
    result.append(x * x)
""",
    number=10_000,
)

三十、性能分析

1. cProfile

import cProfile


def main() -> None:
    result = sum(
        number ** 2
        for number in range(
            1_000_000
        )
    )


cProfile.run(
    "main()"
)

也可以在终端运行：

python -m cProfile script.py

按照累计耗时排序：

python -m cProfile -s cumulative script.py

优化前应先分析，不要凭感觉优化。

2. 性能优化原则

通常优先级：

选择更合理的算法
选择更合适的数据结构
减少不必要的重复计算
使用生成器减少内存
使用内置函数
使用缓存
最后才考虑细节优化

例如，成员判断：

code in code_list

列表平均需要线性查找。

集合：

code in code_set

通常更快。

三十一、内存优化

1. 生成器代替列表

total = sum(
    number ** 2
    for number in range(
        10_000_000
    )
)

生成器不会一次性创建巨大列表。

2. slots

@dataclass(slots=True)
class Stock:
    code: str
    price: float

大量对象时可以减少内存。

3. 避免不必要复制

new_list = old_list[:]

会复制整个列表。

只读使用时，不一定需要复制。

4. 及时释放大对象

del large_data

通常 Python 会自动管理内存，但大型临时数据不再使用时可以主动删除引用。

三十二、路径处理

现代 Python 推荐使用 pathlib。

from pathlib import Path

当前目录：

current_dir = Path.cwd()

用户目录：

home_dir = Path.home()

组合路径：

file_path = (
    Path("data")
    / "stocks.json"
)

判断：

file_path.exists()
file_path.is_file()
file_path.is_dir()

创建目录：

file_path.parent.mkdir(
    parents=True,
    exist_ok=True,
)

查找文件：

for file_path in Path(
    "data"
).glob("*.json"):
    print(file_path)

递归查找：

for file_path in Path(
    "data"
).rglob("*.json"):
    print(file_path)

三十三、JSON 高级处理

import json

格式化输出：

text = json.dumps(
    data,
    ensure_ascii=False,
    indent=2,
)

自定义对象序列化：

from dataclasses import (
    asdict,
    is_dataclass,
)


def json_default(value):
    if is_dataclass(value):
        return asdict(value)

    if isinstance(value, Path):
        return str(value)

    raise TypeError(
        f"无法序列化："
        f"{type(value).__name__}"
    )

使用：

text = json.dumps(
    data,
    ensure_ascii=False,
    indent=2,
    default=json_default,
)

三十四、测试技巧

1. assert

def add(
    a: int,
    b: int,
) -> int:
    return a + b


assert add(1, 2) == 3

assert 适合开发测试，不应代替业务参数验证。

2. unittest

import unittest


class TestAdd(
    unittest.TestCase
):
    def test_positive_numbers(
        self,
    ) -> None:
        self.assertEqual(
            add(1, 2),
            3,
        )

    def test_negative_numbers(
        self,
    ) -> None:
        self.assertEqual(
            add(-1, -2),
            -3,
        )


if __name__ == "__main__":
    unittest.main()

3. 测试异常

class TestStockCode(
    unittest.TestCase
):
    def test_invalid_code(
        self,
    ) -> None:
        with self.assertRaises(
            InvalidStockCodeError
        ):
            validate_code("ABC")

4. 测试原则

重点测试：

正常输入
边界输入
空数据
错误类型
异常情况
极端数据

函数越纯粹，越容易测试。

三十五、依赖注入

不推荐函数内部直接创建所有依赖：

class StockService:
    def get_stock(self):
        database = Database()
        return database.query()

推荐从外部传入：

class StockService:
    def __init__(
        self,
        repository,
    ) -> None:
        self.repository = repository

    def get_stock(
        self,
        code: str,
    ):
        return self.repository.find(
            code
        )

优点：

更容易测试
更容易替换数据库
减少模块耦合
更容易模拟依赖

三十六、单分派函数

根据第一个参数的类型选择不同实现。

from functools import singledispatch


@singledispatch
def serialize(value) -> str:
    raise TypeError(
        f"不支持类型："
        f"{type(value).__name__}"
    )

注册整数：

@serialize.register
def _(
    value: int,
) -> str:
    return str(value)

注册列表：

@serialize.register
def _(
    value: list,
) -> str:
    return ",".join(
        map(str, value)
    )

使用：

print(serialize(100))
print(serialize([1, 2, 3]))

三十七、枚举

固定选项适合使用 Enum。

from enum import Enum


class Direction(Enum):
    BUY = "buy"
    SELL = "sell"

使用：

direction = Direction.BUY

print(direction)
print(direction.value)

判断：

if direction is Direction.BUY:
    print("买入")

1. StrEnum

Python 3.11 以上支持：

from enum import StrEnum


class Direction(StrEnum):
    BUY = "buy"
    SELL = "sell"

StrEnum 的成员可以更自然地作为字符串使用。

2. auto

from enum import (
    Enum,
    auto,
)


class Status(Enum):
    PENDING = auto()
    RUNNING = auto()
    FINISHED = auto()

三十八、描述符

描述符是实现属性管理的底层机制。

class PositiveNumber:
    def __set_name__(
        self,
        owner,
        name,
    ) -> None:
        self.private_name = (
            f"_{name}"
        )

    def __get__(
        self,
        instance,
        owner,
    ):
        if instance is None:
            return self

        return getattr(
            instance,
            self.private_name,
        )

    def __set__(
        self,
        instance,
        value,
    ) -> None:
        if value < 0:
            raise ValueError(
                "数值不能小于零"
            )

        setattr(
            instance,
            self.private_name,
            value,
        )

使用：

class Stock:
    price = PositiveNumber()

    def __init__(
        self,
        price: float,
    ) -> None:
        self.price = price

描述符适合：

属性校验
ORM 字段
框架设计
属性代理

普通业务代码通常使用 property 更简单。

三十九、工程化项目结构

推荐结构：

stock_project/
├── pyproject.toml
├── README.md
├── src/
│   └── stock_project/
│       ├── __init__.py
│       ├── models.py
│       ├── services.py
│       ├── repositories.py
│       ├── exceptions.py
│       └── main.py
└── tests/
    ├── test_models.py
    └── test_services.py

职责示例：

models.py       数据模型
services.py     业务逻辑
repositories.py 数据读写
exceptions.py   自定义异常
main.py         程序入口
tests/          自动化测试

四十、pyproject.toml

现代 Python 项目通常使用 pyproject.toml。

[project]
name = "stock-project"
version = "0.1.0"
description = "股票数据处理项目"
requires-python = ">=3.12"
dependencies = []

[project.optional-dependencies]
dev = [
    "pytest",
    "mypy",
    "ruff",
]

安装开发依赖：

python -m pip install -e ".[dev]"

四十一、综合案例：股票数据处理管道

定义数据模型：

from dataclasses import dataclass


@dataclass(
    slots=True,
    frozen=True,
)
class Stock:
    code: str
    name: str
    price: float
    industry: str

    def __post_init__(self) -> None:
        clean_code = (
            self.code.strip()
        )

        if (
            len(clean_code) != 6
            or not clean_code.isdigit()
        ):
            raise InvalidStockCodeError(
                self.code
            )

        if self.price < 0:
            raise InvalidPriceError(
                str(self.price)
            )

        object.__setattr__(
            self,
            "code",
            clean_code,
        )

        object.__setattr__(
            self,
            "name",
            self.name.strip(),
        )

解析数据：

def parse_stock(
    data: dict[str, object],
) -> Stock:
    try:
        return Stock(
            code=str(data["code"]),
            name=str(data["name"]),
            price=float(data["price"]),
            industry=str(
                data.get(
                    "industry",
                    "未知",
                )
            ),
        )
    except KeyError as error:
        raise StockError(
            f"缺少字段："
            f"{error.args[0]}"
        ) from error

批量解析：

def parse_stocks(
    records: list[
        dict[str, object]
    ],
):
    for index, record in enumerate(
        records,
        start=1,
    ):
        try:
            yield parse_stock(record)
        except StockError as error:
            logger.warning(
                "第 %s 条数据无效：%s",
                index,
                error,
            )

按行业分组：

from collections import defaultdict


def group_by_industry(
    stocks,
) -> dict[str, list[Stock]]:
    groups: defaultdict[
        str,
        list[Stock],
    ] = defaultdict(list)

    for stock in stocks:
        groups[
            stock.industry
        ].append(stock)

    return dict(groups)

统计：

from collections import Counter


def count_industries(
    stocks,
) -> Counter[str]:
    return Counter(
        stock.industry
        for stock in stocks
    )

筛选器：

from collections.abc import (
    Callable,
)


type StockFilter = Callable[
    [Stock],
    bool,
]


def filter_stocks(
    stocks,
    condition: StockFilter,
):
    for stock in stocks:
        if condition(stock):
            yield stock

主板筛选条件：

def is_main_board(
    stock: Stock,
) -> bool:
    return stock.code.startswith(
        ("00", "60")
    )

价格筛选器工厂：

def create_price_filter(
    min_price: float,
    max_price: float | None = None,
) -> StockFilter:
    def condition(
        stock: Stock,
    ) -> bool:
        if stock.price < min_price:
            return False

        if (
            max_price is not None
            and stock.price > max_price
        ):
            return False

        return True

    return condition

完整处理：

def process_records(
    records: list[
        dict[str, object]
    ],
) -> list[Stock]:
    stocks = parse_stocks(records)

    main_board_stocks = filter_stocks(
        stocks,
        is_main_board,
    )

    price_filter = (
        create_price_filter(
            min_price=10,
            max_price=1000,
        )
    )

    filtered_stocks = filter_stocks(
        main_board_stocks,
        price_filter,
    )

    return sorted(
        filtered_stocks,
        key=lambda stock: (
            stock.price
        ),
        reverse=True,
    )

这个案例综合使用了：

数据类
不可变对象
生成器
自定义异常
异常链
日志
类型别名
高阶函数
闭包
惰性处理
排序

四十二、容易被滥用的高级技巧

1. 过长推导式

不推荐：

result = [
    transform(item)
    for group in groups
    if group
    for item in group
    if validate(item)
    if another_condition(item)
]

逻辑复杂时改用普通循环。

2. 过度使用 Lambda

不推荐：

lambda x: (
    complicated_expression_1(x)
    if condition(x)
    else complicated_expression_2(x)
)

应改成命名函数。

3. 过度使用继承

为了复用几行代码建立多层继承，通常得不偿失。

优先考虑组合。

4. 过度使用装饰器

装饰器层数太多，会让函数实际行为难以追踪。

5. 为异步而异步

普通计算任务使用 asyncio 不会自动提升性能。

6. 过早优化

代码还没有性能问题时，不要牺牲可读性换取微小性能提升。

正确顺序：

先写正确
再写清晰
进行测试
测量性能
最后优化

四十三、学习顺序建议

第一阶段：

解包
推导式
排序
collections
itertools

第二阶段：

迭代器
生成器
闭包
高阶函数
装饰器

第三阶段：

上下文管理器
特殊方法
数据类
模式匹配

第四阶段：

类型提示
自定义异常
日志
测试
项目结构

第五阶段：

多线程
多进程
asyncio
性能分析
缓存

四十四、高级 Python 核心原则

原则一：可读性优先

高级语法不是为了少写几行代码，而是为了更准确地表达意图。

原则二：选择正确的数据结构

频繁成员判断 → set
先进先出     → deque
计数         → Counter
自动默认值   → defaultdict
大量数据     → generator
固定数据模型 → dataclass

原则三：避免共享可变状态

共享状态会增加：

并发问题
调试难度
测试难度
维护成本

原则四：优先使用标准库

Python 标准库已经提供了大量成熟工具：

collections
itertools
functools
pathlib
contextlib
concurrent.futures
asyncio
logging
dataclasses
typing

原则五：先测量，再优化

不要猜哪里慢
使用 timeit
使用 cProfile
找到真正瓶颈
再进行优化

原则六：高级不等于复杂

真正成熟的 Python 代码通常具有以下特点：

函数短
职责清晰
类型明确
异常明确
数据流清晰
副作用较少
容易测试
容易修改

高级 Python 的最终目标不是写出别人看不懂的代码，而是：

用更少的复杂度，解决更复杂的问题。