Python 为金融数据分析提供了丰富的内置数据结构和灵活的编程范式。本章系统介绍元组、列表、字典和集合，以及函数式编程的基本概念。
学习目标
完成本章后，你应能：
if、for、推导式及生成器完成小规模金融记录处理；目标—活动—核心习题/答案映射
| 正式目标 | 正文活动 | 核心评价证据 |
|---|---|---|
| 选择内置数据结构并解释复杂度 | 小节 2.2 与 小节 2.3 的性质比较 | 习题 2.1—2.3 及答案中的列表、字典、集合选择与操作 |
| 使用控制流、函数、推导式和生成器 | 小节 2.6、小节 2.5 与 小节 2.6.5 的逐步活动 | 习题 2.1、2.4、2.7 及完整答案 |
| 判断切片、浅复制、嵌套可变对象和哈希键 | 小节 2.2.3、列表 2.1 与 小节 2.2.4.4 的预测—验证活动 | 习题 2.1(e)—(f) 及答案中的原地修改与可哈希性断言 |
| 用上下文管理器完成 CSV 往返 | 小节 2.8 的字段解析和写回活动 | 习题 2.5 及答案中从教师提供 CSV 开始的标准库解析、类型转换和写回记录 |
| 计算真实行情收益与波动率 | 收益率函数和真实长三角公司行情示例 | 习题 2.6 及答案中从教师提供列表开始的循环、函数、标准库统计和极值查找 |
前置知识要求
if、for、函数参数和返回值本章考核边界
核心习题 2.1—2.7 的学生作答与评分只使用本章讲授的内置数据结构、控制流、函数、生成器、标准库和文件操作。为保持案例来自真实中国市场，习题 2.5、2.6 前各有一段教师提供的 Pandas 数据准备脚手架；学生只需运行并核对它产生的 CSV 或“日期—价格”列表，该脚手架的 Pandas/HDF5 语句不要求解释、修改或默写，也不计分。类、dataclass、Pandas 时间序列和策略回测依赖尚未建立的先修知识，因此不进入本章正文或考核；本章结尾只说明后续学习接口。
本章讨论 Python 语言内置的功能，这些功能会在后续金融和经济数据分析中反复使用。Pandas 和 NumPy 面向更大规模的表格与数组计算，但它们仍需与 Python 的内置对象、函数和控制流协同工作。
本节从可直接运行的最小对象开始，依次比较元组、列表、字典和集合。元组与列表保存有序记录，字典保存键值映射，集合用于去重和成员检查；随后再用推导式组织重复的数据变换。
在金融数据处理中，不同的数据结构各有其适用场景。例如，元组（tuple）适合存储不会改变的交易记录；列表（list）适合动态收集行情数据；字典（dict）适合维护股票代码到价格的映射；集合（set）适合执行去重和成员检查。理解这些结构的性质，可以帮助你按数据语义选择容器并预测操作成本。
元组 (tuple) 是一个固定长度、不可变的 Python 对象序列，一旦被赋值，就不能被更改。这种不可变性并非限制，而是一种特性；它保证了一条数据记录，例如股票交易的价格、数量和时间戳，能够保持恒定，不会被意外修改。创建元组最简单的方法是用括号包裹一个逗号分隔的值序列。
下面我们创建一个元组来表示某只股票某个交易日的 OHLC 数据（开盘价、最高价、最低价、收盘价）：
(10.5, 12.0, 10.2, 11.5)Python 直接输出元组的全部内容。四个数字依次对应开盘价 10.5、最高价 12.0、最低价 10.2 和收盘价 11.5；固定顺序使这条题设记录可以按位置读取。
在许多情况下，括号可以省略：
(10.5, 12.0, 10.2, 11.5)你可以通过调用 tuple() 构造函数将任何序列或迭代器转换为元组。这对于“冻结”一个列表的内容非常有用。
('600276', 45.0)
('0', '0', '2', '4', '1', '5')元素可以通过方括号 [] 访问，就像在大多数其他编程语言中一样。按照计算机科学的惯例，Python 中的序列是0索引的。
当你在更复杂的表达式中定义元组时，通常需要将值用括号括起来，例如下面这个创建元组的元组（嵌套元组）的例子：
(('600276', 45.0), ('002415', 32.0))关于“不可变性 (Immutability)”的辨析
元组的“不可变性”指的是元组本身所包含的对象的引用是不可更改的，即你不能将元组某个位置上的对象替换成另一个对象。然而，如果元组中的某个对象本身是“可变的”（比如一个列表），那么这个对象的内容是可以被修改的。这个特性在处理复杂数据结构时非常重要。
尽管存储在元组中的对象本身可能是可变的，但一旦元组被创建，就不可能修改每个槽位中存储的对象：
--------------------------------------------------------------------------- TypeError Traceback (most recent call last) Cell In[6], line 3 1 trade_tuple = tuple(['600276', 100]) # 用恒瑞医药代码与题设股数演示不可变交易记录 2 # 下面这行会引发 TypeError，因为你不能给元组的一个槽位赋一个新的对象 ----> 3 trade_tuple[1] = 200 # 尝试修改元组中的交易数量，将触发TypeError TypeError: 'tuple' object does not support item assignment
但是，如果元组内部的一个对象是可变的，比如一个列表，你可以就地修改它。这是一个至关重要的区别。
('Fund_A', ['600276', '002415', '600104'], True)你可以使用 + 运算符连接元组以产生更长的元组。请注意，这将创建一个新的元组；它不会修改原始元组。
('2024-01-01', 'BUY', '600276', 100, 45.0)将元组乘以一个整数，效果是连接该元组的多个副本。对象本身不会被复制，只有它们的引用会被复制。在处理大型数据集时，这对内存管理来说是一个重要的细节。
('BUY', 'SELL', 'BUY', 'SELL', 'BUY', 'SELL', 'BUY', 'SELL')元组一个非常方便的特性是解包 (unpacking)。如果你试图将一个值赋给一个元组式的变量表达式，Python 会尝试解包等号右侧的值：
high_px is: 182.0即使是带有嵌套元组的序列也可以被解包，这对于迭代结构化数据非常有用。
symbol is: 600276这个功能为交换变量值提供了一种优雅的方式，这个任务在许多其他语言中需要一个临时变量。
变量交换的底层机制
bid_price, ask_price = ask_price, bid_price 这种简洁语法的背后原理是元组的创建和解包。
ask_price, bid_price，这会创建一个临时的、未命名的元组。整个过程是原子性的，确保了交换的正确性，即使变量名相同也不会出错。
Before swap: bid=10.0, ask=10.2
After swap: bid=10.2, ask=10.0变量解包的一个常见用途是迭代元组或列表的序列：
symbol=600276, qty=100, price=45.0
symbol=002415, qty=50, price=32.0
symbol=600104, qty=200, price=15.0有时，你可能想从序列的开头“摘取”几个元素，并收集其余的元素。*rest 语法可以用于此目的。在经济时间序列分析中，这对于将最新的数据点与历史序列分开非常有用。
# 最近5日的收盘价 (从最新到最旧)
latest_prices = 105, 104, 103, 102, 100 # 最近5日收盘价（从最新到最旧）
today, yesterday, *history = latest_prices # 前两个赋给today和yesterday，其余收集到history列表
print(f'today = {today}') # 输出今日收盘价：105
print(f'yesterday = {yesterday}') # 输出昨日收盘价：104
print(f'history = {history}') # 输出历史价格列表：[103, 102, 100]today = 105
yesterday = 104
history = [103, 102, 100]按照惯例，许多 Python 程序员在解包时会使用下划线 (_) 来表示不想要的变量。
由于元组的大小和内容不能被修改，它的实例方法非常少。一个特别有用的方法是 count（也适用于列表），它计算一个值出现的次数。
了解了不可变的元组之后，我们来看它的”可变孪生兄弟”——列表 (list)。与元组相反，列表的长度是可变的，其内容可以就地修改。它们是可变的 (mutable)。在数据分析中，你可能会用列表来累积模拟结果，或者存放一个正在增量更新的经济时间序列数据。你可以用方括号 [] 或 list 类型函数来定义它们：
['600276', '002415', '600104']['600276', '601018', '600104']列表和元组在语义上相似，在许多函数中可以互换使用。list 函数在数据处理中经常被用来物化 (materialize) 一个迭代器或生成器表达式（我们稍后会探讨这些概念）。
range(0, 10)[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]可以使用 append 方法将元素添加到列表的末尾：
['600276', '601018', '600104', '002230']使用 insert 可以在列表的特定位置插入一个元素。插入索引必须在 0 和列表长度（包含）之间。
['600276', '002142', '601018', '600104', '002230']append 与 insert 的性能考量
在处理大规模金融数据集时，理解基本操作的计算成本至关重要：
append(value): 将元素添加到列表末尾的摊销时间复杂度为 O(1)。多数追加直接写入预留容量；容量不足时必须申请更大存储并搬移已有元素，因此触发扩容的那一次追加最坏为 O(n)。所以“摊销 O(1)”描述的是一串追加操作的平均成本，并非每次调用都与列表大小无关。insert(i, value): 在列表的任意位置 i 插入元素是一个 O(n) 操作。这是因为插入点之后的所有元素都需要向右移动一位。对于大型历史数据集，频繁使用 insert 会导致显著的性能瓶颈。因此，在构建大型资产列表时，应优先使用 append。如果确实需要在序列两端进行高效操作，可以考虑使用 collections.deque。
insert 的逆操作是 pop，它会移除并返回特定索引处的元素：
['600276', '002142', '600104', '002230']可以使用 remove 按值移除元素，它会找到第一个这样的值并从列表中移除：
移除前: ['600276', '002142', '600104', '002230', '600276']
移除后: ['002142', '600104', '002230', '600276']你可以使用 in 关键字检查列表中是否包含某个值：
not 关键字可以用来否定 in：
列表搜索的计算复杂度
检查一个值是否存在于列表中（value in ticker_list）是一个 O(n) 操作。Python 必须从头到尾对列表进行线性扫描，直到找到匹配项或结束。
相比之下，字典（dict）和集合（set）使用哈希表，其成员查找在哈希分布良好、负载因子受控时具有典型/期望 O(1) 复杂度。若大量键碰撞或出现退化探测序列，单次查找最坏可达 O(n)。在需要频繁检查某个股票是否在持仓池（portfolio pool）中时，集合通常仍比线性扫描列表更合适，但不能把期望界误写成无条件最坏界。
与元组类似，用 + 将两个列表相加会连接它们：
[10.5, None, 11.2, 12.0, 11.8, (2, 3)]如果你已经定义了一个列表，你可以使用 extend 方法向其追加多个元素：
[10.5, None, 11.2, 12.0, 11.8, (2, 3)]请注意，通过加法进行列表连接是一个相对昂贵的操作，因为必须创建一个新列表并将对象复制过去。使用 extend 向现有列表追加元素，尤其是在构建大列表时，通常是更好的选择，因为它会就地修改列表。
你可以通过调用列表的 sort 方法对其进行就地排序（不创建新对象）：
[1, 2, 3, 5, 7]sort 有一些偶尔会派上用场的选项。其中之一是传递一个次要排序键 (key)——即一个函数，它为每个元素生成一个用于排序的值。例如，我们可以按字符串的长度对一个字符串集合进行排序：
['He', 'saw', 'six', 'small', 'foxes']稍后，我们将学习 sorted 函数，它可以从一个通用序列生成一个排好序的副本。
你可以通过使用切片表示法来选择大多数序列类型的部分，其基本形式是 start:stop，传递给索引运算符 []。
[2, 5, 1, 3]切片也可以被赋值：
[7, 2, 5, 'a', 'b', 6, 8, 9]虽然 start 索引处的元素被包含在内，但 stop 索引处的元素不被包含，因此结果中的元素数量是 stop - start。start 或 stop 都可以省略，此时它们分别默认为序列的开头或结尾：
[7, 2, 5, 'a', 'b']
['a', 'b', 6, 8, 9]负数索引表示从序列末尾开始切片：
['b', 6, 8, 9]
[5, 'a', 'b', 6]切片的语义需要一些时间来适应，特别是如果你之前使用过 R 或 MATLAB。图 2.1 提供了一个有用的图示，说明了使用正整数和负整数进行切片。在图中，索引显示在“箱格”的边缘，以帮助说明切片选择的开始和停止位置。
import matplotlib.pyplot as plt # 使用对象式绘图把切片边界与元素位置放在同一坐标系
import numpy as np # 为切片区间的连续边界位置提供数值坐标
# 数据和设置
s = list('python') # 将字符串'python'拆分为字符列表
n = len(s) # 计算字符列表的长度
fig, ax = plt.subplots(figsize=(10, 3)) # 使用宽画布给正负索引与边界标签留出垂直空间
ax.set_ylim(-1, 1.5) # 同时容纳元素上方正索引和下方负索引
ax.set_xlim(-0.5, n + 0.5) # 在首尾边界外留白，避免端点标签被裁切
# 绘制字符和框
for i, char in enumerate(s): # 遍历每个字符及其索引
ax.text(i + 0.5, 0.5, char, ha='center', va='center', fontsize=20, fontfamily='monospace') # 在格子中心绘制字符
rect = plt.Rectangle((i, 0.3), 1, 0.4, fill=None, edgecolor='black', linewidth=1) # 创建字符格子的矩形边框
ax.add_patch(rect) # 将矩形添加到坐标轴
# 绘制正向索引 (元素上方)
for i in range(n): # 遍历每个位置绘制正向索引
ax.text(i + 0.5, 1.1, str(i), ha='center', va='bottom', fontsize=12, color='blue') # 在字符上方显示正向索引（蓝色）
# 绘制负向索引 (元素下方)
for i in range(n): # 遍历每个位置绘制负向索引
ax.text(i + 0.5, -0.1, str(i-n), ha='center', va='top', fontsize=12, color='red') # 在字符下方显示负向索引（红色）
# 绘制 "bin edge" 索引
for i in range(n + 1): # 遍历每个边界位置绘制切片索引
ax.text(i, 0.9, str(i), ha='center', va='bottom', fontsize=10, color='gray') # 在格子边界上方显示正向切片索引（灰色）
ax.axvline(i, ymin=0.35, ymax=0.65, color='gray', linestyle=':') # 绘制格子边界的灰色虚线
for i in range(n + 1): # 遍历每个边界位置绘制负向切片索引
ax.text(i, 0.1, str(i-n) if i != 0 else str(i-n)+'/0', ha='center', va='top', fontsize=10, color='gray') # 在格子边界下方显示负向切片索引# 添加示例切片说明
ax.annotate('seq[1:5]', xy=(1, -0.5), xytext=(1, -0.9), # 标注seq[1:5]切片范围的起点
arrowprops=dict(arrowstyle='-', connectionstyle='angle3,angleA=0,angleB=90', color='green'), # 设置绿色箭头连接样式
ha='left', va='top', color='green', fontsize=12) # 绿色文字、左对齐
ax.annotate('', xy=(5, -0.5), xytext=(5, -0.9), # 标注seq[1:5]切片范围的终点
arrowprops=dict(arrowstyle='-', connectionstyle='angle3,angleA=0,angleB=90', color='green')) # 同样的绿色箭头样式
ax.add_patch(plt.Rectangle((1, -0.55), 4, 0.05, color='green', alpha=0.3)) # 绘制绿色半透明矩形表示seq[1:5]的选取区域
ax.annotate('seq[-4:]', xy=(n-4, -0.6), xytext=(n-4, -1.2), # 标注seq[-4:]切片范围的起点
arrowprops=dict(arrowstyle='-', connectionstyle='angle3,angleA=0,angleB=90', color='purple'), # 设置紫色箭头连接样式
ha='left', va='top', color='purple', fontsize=12) # 紫色文字、左对齐
ax.annotate('', xy=(n, -0.6), xytext=(n, -1.2), # 标注seq[-4:]切片范围的终点
arrowprops=dict(arrowstyle='-', connectionstyle='angle3,angleA=0,angleB=90', color='purple')) # 同样的紫色箭头样式
ax.add_patch(plt.Rectangle(((n-4), -0.65), 4, 0.05, color='purple', alpha=0.3)) # 绘制紫色半透明矩形表示seq[-4:]的选取区域
# 图中自带边界与索引标记，关闭默认坐标轴以免形成第二套刻度
ax.axis('off') # 只保留教学所需的自定义索引系统
plt.show() # 输出最终切片示意图，供正文比较左闭右开区间在第二个冒号后还可以使用一个 step，例如，可以用来取每隔一个元素：
[1000, 1800, 1500, 4000]一个巧妙的用法是传递 -1，这有一个有用的效果，就是反转一个列表或元组：
掌握列表后，我们转向按键查找的字典 (dictionary)。如果把列表理解为按位置编号的储物格，字典则相当于按标签查找的储物格。字典也称为哈希映射 (hash map) 或关联数组 (associative array)，用于把唯一标识符（如国家 ISO 代码或证券代码）映射到相应记录。
创建字典的一种方法是使用花括号 {} 和冒号来分隔键和值：
{'symbol': '600276', 'price': 45.0, 'volume': 50000}你可以使用与访问列表或元组元素相同的语法来访问、插入或设置元素：
{'symbol': '600276', 'price': 182.0, 'volume': 50000}
182.0你可以使用与检查列表或元组是否包含某个值相同的语法来检查字典是否包含某个键：
你可以使用 del 关键字或 pop 方法（它会同时返回值并删除键）来删除值：
stock_info['temp'] = 'to be deleted' # 新增一个临时键值对
stock_info['market'] = 'A股市场' # 新增市场信息键值对
print(f'del 之前: {stock_info}') # 显示删除操作前的字典
del stock_info['temp'] # 使用del关键字删除指定键值对
print(f'del 之后: {stock_info}') # 显示del删除后的字典
ret = stock_info.pop('market') # 用pop方法删除键'market'并返回其值
print(f'弹出的值: {ret}') # 显示pop返回的被删除值
print(f'pop 之后: {stock_info}') # 显示pop后的字典del 之前: {'symbol': '600276', 'price': 182.0, 'volume': 50000, 'temp': 'to be deleted', 'market': 'A股市场'}
del 之后: {'symbol': '600276', 'price': 182.0, 'volume': 50000, 'market': 'A股市场'}
弹出的值: A股市场
pop 之后: {'symbol': '600276', 'price': 182.0, 'volume': 50000}keys 和 values 方法分别提供字典键和值的迭代器。
['symbol', 'price', 'volume']
['600276', 182.0, 50000]如果你需要同时迭代键和值，可以使用 items 方法来迭代键值对（作为2元组）：
你可以使用 update 方法将一个字典合并到另一个字典中。这将就地修改字典，因此传递给 update 的数据中任何已存在的键，其旧值都将被丢弃。
{'symbol': '600276', 'price': 185.0, 'volume': 50000, 'pe_ratio': 30.5}通常情况下，你可能会得到两个序列，并希望在字典中将它们按元素配对。zip 函数与 dict 构造函数结合使用，可以轻松实现这一点。
下面的逻辑很常见：
字典的 get 方法提供了一种更简洁的写法。如果键不存在，get 将返回 None 或一个指定的默认值。
0.0
183.05一个常见的用例是让字典的值是其他集合，比如列表。例如，按首字母对一个单词列表进行分类。
words = ['apple', 'bat', 'bar', 'atom', 'book'] # 待分组的单词列表
by_letter = {} # 创建空字典用于存储分组结果
for word in words: # 逐项读取apple等五个词，显式演示首次建组与后续追加两条分支
letter = word[0] # 取单词的首字母作为分组键
if letter not in by_letter: # 如果该首字母尚未在字典中
by_letter[letter] = [word] # 创建新的列表存入第一个单词
else: # 如果首字母已存在
by_letter[letter].append(word) # 将单词追加到已有列表
by_letter # 查看按首字母分组的结果{'a': ['apple', 'atom'], 'b': ['bat', 'bar', 'book']}setdefault 方法可以简化这个工作流程。前面的 for 循环可以重写为：
{'a': ['apple', 'atom'], 'b': ['bat', 'bar', 'book']}内置的 collections 模块有一个有用的类 defaultdict，这使得分组更加容易。
{'a': ['apple', 'atom'], 'b': ['bat', 'bar', 'book']}在深入理解字典键的限制之前，我们需要从数学角度理解哈希表（Hash Table）的工作原理。
语言契约与抽象哈希模型
Python 的语言契约首先约束“键是否可哈希”，但不规定字典必须采用哪一种冲突解决算法。对象 \(k\) 可作为字典键，至少需要满足两条语义条件：同一个对象在其生命周期内的哈希值保持稳定；若两个对象相等，则它们的哈希值必须相等，即
\[ x = y \Longrightarrow \operatorname{hash}(x)=\operatorname{hash}(y). \tag{2.1}\]
逆命题不成立：不同对象允许有相同哈希值，这正是哈希冲突。不可变性通常有助于满足稳定性，但“不可变”不是充分条件，例如包含列表的元组仍不可哈希；反之，自定义类型是否可哈希由其 __hash__ 与 __eq__ 契约共同决定。
为了分析复杂度，可以另行引入一个随机哈希族。设 \(\mathcal{H}\) 是从键空间 \(U\) 到 \(m\) 个桶 \(\{0,\ldots,m-1\}\) 的函数族，在建表时随机抽取 \(H\sim\mathcal{H}\)。简单均匀哈希假设写为：对任意固定键 \(k\) 和任意桶 \(j\)，
\[ \Pr\{H(k)=j\}=\frac{1}{m}, \qquad j=0,\ldots,m-1. \tag{2.2}\]
式 式 2.2 给出本节后续实现与解释采用的数学关系。
若进一步假设不同键的桶位置相互独立，则有
\[ \Pr\{H(k_1)=i,H(k_2)=j\}=\frac{1}{m^2}, \qquad k_1\ne k_2. \tag{2.3}\]
式 式 2.3 给出本节后续实现与解释采用的数学关系。
这里的概率来自随机选择 \(H\)，而不是说一次程序运行中的确定函数会“随机改变”。真实实现通常只能近似这些理想条件，因此它们是复杂度分析模型，而非 Python 对每个输入作出的保证。
冲突解决：语言保证与 CPython 实现的边界
当 \(k_1\ne k_2\) 但 \(H(k_1)=H(k_2)\) 时发生冲突。Python 语言层面只保证字典映射的可观察语义，并未规定使用链地址、开放寻址或某个探测公式。CPython 的字典实现采用开放寻址，但表布局、扰动方式和探测序列属于具体版本的实现细节，可能随版本改变；因此不能把通用二次探测公式当作 Python 语言契约。业务代码应依赖键相等、插入、删除和查找的公开语义，而不依赖某个槽位位置。
成功查找、失败查找与最坏情况
设开放寻址表中已有 \(n\) 个键、容量为 \(m\)、负载因子为 \(\alpha=n/m<1\)。在简单均匀哈希的理想模型下，失败查找的期望探测次数上界为
\[ E[C_{\mathrm{fail}}]\leq\frac{1}{1-\alpha}, \tag{2.4}\]
式 式 2.4 给出本节后续实现与解释采用的数学关系。
成功查找的期望探测次数上界则为
\[ E[C_{\mathrm{success}}]\leq\frac{1}{\alpha}\ln\!\left(\frac{1}{1-\alpha}\right). \tag{2.5}\]
式 式 2.5 给出本节后续实现与解释采用的数学关系。
当 \(\alpha\) 由扩容策略控制在远离 1 的范围时，两者关于 \(n\) 都是期望 \(O(1)\)；但两条公式并不相同，也不能脱离均匀哈希假设使用。若大量键冲突或遭遇对抗性输入，单次查找最坏可退化为 \(O(n)\)。所以教材中“字典查找为常数时间”应理解为通常情况下的摊销期望结论，而不是无条件最坏界。
虽然字典的值可以是任何 Python 对象，但键必须是可哈希对象。字符串、数字以及所有元素均可哈希的元组是常用键类型；“优先选择不发生值变化的标识符类型”是稳健的工程建议，却不是可哈希性的充分必要定义。
关键概念：可哈希性 (Hashability)
字典在内部使用哈希表来存储数据，使键查找通常具有较低成本。Python 通过对象的哈希协议取得整数哈希值，再结合相等性比较定位键；协议作用于可哈希对象，而不是笼统地作用于所有不可变对象。
为了让系统可靠工作，一个关键要求是：一个对象的哈希值在其生命周期内必须保持稳定，并满足 式 2.1 的相等—哈希一致性。
因此，金融分析中的唯一标识符（如 order_book_id）通常选择字符串或只含可哈希元素的元组：它们既表达稳定标识，也能直接作为键。真正的必要条件仍是满足哈希契约；若只需要普通列而不作为键，则数据类型还应依据字段语义、缺失规则和存储成本决定。
你可以使用 hash 函数检查一个对象是否是可哈希的：
-3345294237905129588
-9209053662355515447--------------------------------------------------------------------------- TypeError Traceback (most recent call last) Cell In[50], line 4 2 print(hash((1, 2, (2, 3)))) # 核验嵌套元组的每个元素均可哈希，整体因而可哈希 3 # 下面这行会失败，因为Python内置list没有提供可用的哈希值。 ----> 4 hash((1, 2, [2, 3])) # 元组包含不可哈希列表，整体无法满足字典键契约并抛出TypeError TypeError: unhashable type: 'list'
要使用列表作为键，一个选择是将其转换为元组：
学习完字典之后，我们来看 Python 的最后一个核心数据结构——集合 (set)。集合是一个无序的唯一元素集合，其最大特点是自动去除重复元素。如果说字典是带有”标签”的键值容器，那么集合就是一个只存”标签”（键）、不存”内容”（值）的特殊容器。它在成员资格测试、从序列中移除重复项以及数学运算（如并集、交集和差集）方面非常有用。在金融场景中，你可以使用集合来快速确认某只股票是否已在你的持仓池中，或者找到两个投资组合的共同持仓。
集合可以通过两种方式创建：通过 set 函数或通过带有花括号的集合字面量：
set()
{600000, 601009, 600926}注意输出结果只保留 600000、600926 和 601009 各一次——尽管输入包含重复代码，集合会自动去重。这一特性适合清洗证券池中的重复标识。
集合支持数学集合运算。考虑这两个模拟投资组合的示例集合：
这两个集合的并集 (union) 是出现在任一集合中的不同元素的集合。这可以通过 union 方法或 | 二元运算符来计算：
{'600926', '600104', '002415', '600276', '002142', '002230', '601009', '601018'}
{'600926', '600104', '002415', '600276', '002142', '002230', '601009', '601018'}交集 (intersection) 包含同时出现在两个集合中的元素。可以使用 & 运算符或 intersection 方法：
{'600926', '600104', '002142'}
{'600926', '600104', '002142'}表 2.1 列出了常用的集合方法。
| 函数 | 替代语法 | 描述 |
|---|---|---|
a.add(x) |
N/A | 将元素 x 添加到集合 a。 |
a.clear() |
N/A | 将集合 a 重置为空状态。 |
a.remove(x) |
N/A | 从集合 a 中移除元素 x。 |
a.pop() |
N/A | 从集合 a 中移除一个任意元素。 |
a.union(b) |
a \| b |
a 和 b 中的所有唯一元素。 |
a.update(b) |
a \|= b |
将 a 设置为 a 和 b 的并集。 |
a.intersection(b) |
a & b |
a 和 b 中的所有共同元素。 |
a.intersection_update(b) |
a &= b |
将 a 设置为 a 和 b 的交集。 |
a.difference(b) |
a - b |
a 中存在但 b 中不存在的元素。 |
a.difference_update(b) |
a -= b |
将 a 设置为 a 和 b 的差集。 |
a.symmetric_difference(b) |
a ^ b |
存在于 a 或 b 中但不同时存在的元素。 |
a.symmetric_difference_update(b) |
a ^= b |
将 a 设置为对称差集。 |
a.issubset(b) |
<= b |
如果 a 是 b 的子集，则为 True。 |
a.issuperset(b) |
>= b |
如果 a 是 b 的超集，则为 True。 |
a.isdisjoint(b) |
N/A | 如果 a 和 b 没有共同元素，则为 True。 |
所有的逻辑集合运算都有就地 (in-place) 版本，对于非常大的集合，这可能更高效。
就地并集: {'600926', '600104', '002415', '600276', '002142', '002230', '601009', '601018'}
就地交集: {'600926', '600104', '002142'}与字典键一样，集合元素也必须可哈希。若列表内容代表一条固定值记录，可将其转换为元素均可哈希的元组；转换只适用于记录语义本来就不应再变化的场景：
{(1, 2, 3)}你还可以检查子集和超集：
True
True当且仅当集合的内容相等时，它们才相等；顺序无关紧要。
前文已经通过元组、列表、字典和集合完成了创建、修改、查找与去重。现在可以把这些具体行为提升到两个问题：容器向使用者承诺什么操作，以及具体实现需要付出多少时间与空间成本。本节用于解释前文现象；第一次阅读时只需掌握比较结论，形式化表达可在完成基础练习后复习。
数据抽象与抽象数据类型（Abstract Data Type, ADT）
抽象数据类型（Abstract Data Type, ADT）与具体数据结构处在不同抽象层次。ADT 用“允许保存什么值、允许执行什么操作以及这些操作应满足什么行为”描述接口；列表、动态数组、链表或哈希表等数据结构则说明该接口如何在内存中实现。Python 的 list 是一种具体容器类型，其外部行为可按列表 ADT 理解，而实现采用动态数组；不能把“ADT”和“数据结构”直接当作同义词。
一个 ADT 可以由两个要素描述：
有限列表的值域应写成有序序列族，而不是会消除顺序和重复元素的集合。若元素来自集合 \(E\)，列表值域可写为
\[ D=\bigcup_{n\geq 0}E^n. \]
其中，\((a_1,a_2,a_1)\in E^3\) 与 \((a_1,a_1,a_2)\in E^3\) 是不同列表，重复元素也不会被删除。操作集合可写为 \(\Omega=\{\operatorname{append},\operatorname{remove},\operatorname{sort},\operatorname{slice},\ldots\}\)；每项操作的行为由接口约定，复杂度则取决于具体实现。
从已经执行的操作理解复杂度
前文的索引、成员检查、追加和删除并不具有相同成本。下表给出 CPython 常见实现下需要记住的渐近界；“典型/期望”不是无条件的最坏情况保证。
| 操作 | 列表 | 字典/集合 | 说明 |
|---|---|---|---|
| 访问元素 | 按索引 O(1) | 按键典型/期望 O(1)，最坏 O(n) | 哈希冲突集中时，字典/集合可能退化 |
| 搜索元素 | O(n) | 典型/期望 O(1)，最坏 O(n) | 列表线性扫描；哈希表的常数时间不是无条件最坏界 |
| 插入元素 | 末尾追加摊销 O(1)，扩容时单次最坏 O(n)；中间插入 O(n) | 典型/期望 O(1)，最坏 O(n) | 列表扩容需搬移已有引用，中间插入还需移动后缀元素 |
| 删除元素 | O(n) | 典型/期望 O(1)，最坏 O(n) | 列表需移动后缀元素；哈希碰撞会拉长字典/集合探测序列 |
| 切片 | O(k) | 不适用 | \(k\) 为切片产生的元素数 |
表 表 2.2 解释了为什么反复做成员检查时通常选择集合，而需要保留顺序和重复元素时仍选择列表。数据规模很小时，可读性往往比常数项优化更重要；规模增大后再结合实际输入测量。
可变性、可哈希性与线程安全是三套不同条件
前文已经观察到列表可以原地改变、元组槽位不能重新绑定、字典键需要可哈希对象。这些现象不能被压缩成“不可变对象总是安全”一句话：
在金融数据处理中，选择类型时应分别问三个问题：记录能否原地变化、对象能否满足键的哈希契约、并发访问采用何种同步协议。把证券代码保存为字符串通常方便防止误改并用作键，但这是工程选择，不是“不可变必然线程安全且可哈希”的逻辑定理。
掌握了核心数据结构之后，我们来认识几个 Python 的内置序列函数。这些函数是日常数据处理的”瑞士军刀”，能够大幅简化对列表、元组等序列的常见操作。每一个函数都体现了 Python 的设计哲学：提供简洁、可读的语法来完成常见任务。
在迭代序列时，通常需要跟踪当前项的索引。如果你习惯于 C 或 Java 风格的 for (int i = 0; i < n; i++) 循环，enumerate 函数提供了一种更 Pythonic（即更符合 Python 风格）的解决方案。enumerate 返回一个 (i, value) 元组的序列：
sorted 函数从任何序列的元素中返回一个新的排好序的列表。这与 list.sort() 方法形成对比，后者是就地对列表进行排序。
zip 将多个列表、元组或其他序列的元素“配对”起来，创建一个元组的列表。
[('stock', 'one'), ('bond', 'two'), ('fund', 'three')]zip 可以接受任意数量的序列，它产生的元素数量由最短的序列决定。
[('stock', 'one', False), ('bond', 'two', True)]zip 的一个常见用途是同时迭代多个序列，可能还会与 enumerate 结合使用：
reversed 以相反的顺序迭代序列的元素。
请记住，reversed(range(10)) 返回的是一个遵循迭代器协议的反向迭代器，而不是生成器。它会按需给出元素，因此用 list() 才会把全部结果物化；一般迭代器与由生成器函数或生成器表达式创建的生成器不能混为一谈。
列表推导式 (List comprehensions) 是一个方便且被广泛使用的 Python 语言特性。它们允许你通过一个简洁的表达式，从一个集合中筛选元素并进行转换，从而简洁地形成一个新列表。基本形式是：
从数学角度看，列表推导式是集合概括（Set-Comprehension）思想的程序实现。
集合概括的数学定义
在朴素集合论（Naive Set Theory）中，集合概括定义为：
\[ S = \{x \in A \mid P(x)\} \]
其中： - \(A\) 是源集合 - \(P(x)\) 是关于元素 \(x\) 的性质或谓词 - \(S\) 是满足性质的所有元素的集合
示例：偶数集合的数学表示 \[ E = \{x \in \{1, 2, ..., 100\} \mid (x \mod 2 = 0)\} \]
在 Python 中，这对应于：
映射与过滤的数学表示
更复杂的推导式涉及映射（Mapping）和过滤（Filtering）的组合。
给定： - 源集合 \(A\) - 映射函数 \(f: A \to B\) - 谓词函数 \(P: A \to \{\text{True}, \text{False}\}\)
过滤后映射定义为： \[ R = \{f(x) \mid x \in A \land P(x)\} \]
这在 Python 中直接对应于：
多重集合概括（嵌套推导式）
对于嵌套结构，我们使用笛卡尔积（Cartesian Product）的概念：
\[ A \times B = \{(a, b) \mid a \in A \land b \in B\} \]
Python 的嵌套推导式：
这正是笛卡尔积的实现，时间复杂度为 \(O(|A| \times |B|)\)。
复杂度对比
| 方法 | 时间复杂度 | 空间复杂度 | 说明 |
|---|---|---|---|
| 传统 for 循环 | O(n) | O(n) | 需要预分配列表空间 |
| 列表推导式 | O(n) | O(n) | 更简洁，Python 内部优化 |
| 生成器表达式 | O(n) | 本例额外 O(1) | 惰性产出；不计既有源对象，且实际仍保留迭代状态与引用 |
对于金融数据处理，列表推导式提供了声明式、函数式的编程风格，使代码更易读和维护。 [expr for value in collection if condition]
这等同于以下 for 循环：
if condition 部分是可选的。例如，给定一个字符串列表，我们可以过滤掉长度为2或更短的字符串，并将它们转换为大写：
['BAT', 'CAR', 'DOVE', 'PYTHON']集合和字典推导式是一个自然的扩展。集合推导式看起来与等效的列表推导式相似，只是用花括号代替：
{1, 2, 3, 4, 6}字典推导式看起来像这样： {key-expr: value-expr for value in collection if condition} 例如，我们可以创建一个这些字符串到它们在列表中位置的查找映射：
假设我们有一个包含一些英文和西班牙文名字的列表的列表。
假设我们想得到一个包含所有名字中含有两个或更多”a”的单个列表。我们可以用嵌套的 for 循环来做到这一点，但嵌套列表推导式更简洁。
['Maria', 'Natalia']列表推导式的 for 部分是根据嵌套的顺序排列的。这里是另一个例子，我们将一个元组列表“扁平化”为一个简单的整数列表：
[1, 2, 3, 4, 5, 6, 7, 8, 9]重要的是要区分刚才展示的语法和列表推导式内部的列表推导式，后者会产生一个列表的列表：
学习了数据结构和序列操作之后，我们进入 Python 编程的核心：函数 (Functions)。函数是 Python 中代码组织和复用的主要且最重要的方法。将代码封装进函数，不仅可以避免重复编写相似的逻辑，还能提高代码的可读性和可维护性。一条基本的工程原则是：DRY（Don’t Repeat Yourself）——不要重复自己的代码。根据经验，如果你预计需要重复相同或非常相似的代码超过一次，那么编写一个可复用的函数是值得的。函数用 def 关键字声明，后跟函数名和参数列表：
每个函数可以有位置参数 (positional arguments) 和关键字参数 (keyword arguments)。关键字参数最常用于指定默认值或可选参数。
虽然关键字参数是可选的，但在调用函数时必须指定所有位置参数。主要的限制是关键字参数必须跟在位置参数（如果有的话）之后。
0.06363636363636363
45.0函数可以访问不同命名空间 (namespaces) 中的变量。默认情况下，在函数内部赋的任何变量都赋给了局部命名空间 (local namespace)。局部命名空间在函数被调用时创建，并立即由函数的参数填充。函数结束后，局部命名空间被销毁。 考虑这个函数：
现在，假设我们在函数外部声明了 a。函数可以从更高的作用域访问这个变量。
[0, 1, 2, 3, 4]
[0, 1, 2, 3, 4, 0, 1, 2, 3, 4]对函数作用域之外的变量进行赋值是可能的，但它们必须使用 global 关键字声明。我通常不鼓励使用 global 关键字。
Python 函数一个非常方便的特性是能够返回多个值。
a=5, b=6, c=7这里发生的是，函数实际上只返回一个对象，一个元组，然后这个元组被解包到结果变量中。
一个可能有吸引力的替代方案是返回一个字典：
由于 Python 函数是对象，可以将函数作为参数传递给其他函数。这是函数式编程的基本机制。下面用一组地区名称演示按名称长度排序。
为了使这个列表统一，我们需要去除空白、移除标点符号并统一大小写。
import re # 导入正则表达式模块
def clean_strings(strings): # 定义清洗字符串列表的函数
result = [] # 初始化空列表，用于存放清洗后的字符串
for value in strings: # 遍历待清洗的每个字符串
value = value.strip() # 去除首尾空白字符
value = re.sub('[!#?]', '', value) # 用正则表达式移除标点符号!#?
value = value.title() # 转换为标题格式（每个单词首字母大写）
result.append(value) # 将清洗后的字符串添加到结果列表
return result # 返回清洗后的字符串列表
clean_strings(states) # 调用函数清洗states列表，返回['Alabama', 'Georgia', ...]['Alabama',
'Georgia',
'Georgia',
'Georgia',
'Florida',
'South Carolina',
'West Virginia']一种更灵活、函数式的方法是定义一个你想要应用的操作列表。
def remove_punctuation(value): # 定义移除标点符号的独立函数
return re.sub('[!#?]', '', value) # 返回移除!#?标点后的字符串
clean_ops = [str.strip, remove_punctuation, str.title] # 定义清洗操作列表：去空白→去标点→标题化
def clean_strings(strings, ops): # 定义接受操作列表作为参数的通用清洗函数
result = [] # 初始化空列表存放结果
for value in strings: # 遍历待清洗的每个字符串
for func in ops: # 依次应用操作列表中的每个函数
value = func(value) # 将当前值传入函数，用返回值更新
result.append(value) # 将经过全部操作后的字符串添加到结果
return result # 返回清洗后的字符串列表
clean_strings(states, clean_ops) # 用操作列表清洗states，结果同上但更灵活['Alabama',
'Georgia',
'Georgia',
'Georgia',
'Florida',
'South Carolina',
'West Virginia']你还可以将函数用作其他内置函数的参数，比如 map，它将一个函数应用于一个序列。
Python 支持所谓的匿名 (anonymous) 或 lambda 函数，这是一种编写只包含单个语句的函数的方式，该语句的结果就是返回值。
它们在数据分析中特别方便，因为许多数据转换函数都接受其他函数作为参数。传递一个 lambda 函数通常比编写一个完整的函数声明更清晰。
[8, 14, 2, 4, 18]再举一个例子，假设你想按每个字符串中不同字母的数量对一个字符串集合进行排序：
Python 中的许多对象都支持迭代。这是通过迭代器协议 (iterator protocol) 实现的。例如，迭代一个字典会产生字典的键。
a
b
c当你写 for key in some_dict 时，Python 解释器首先尝试从 some_dict 创建一个迭代器。
<dict_keyiterator object at 0x74a8bdcbaca0>['a', 'b', 'c']生成器 (generator) 是构建新可迭代对象的一种便捷方式。普通函数执行并返回单个结果，而生成器可以通过暂停和恢复执行来返回一系列多个值。要创建一个生成器，使用 yield 关键字而不是 return。
从计算机科学和函数式编程理论角度理解生成器需要掌握惰性求值的核心概念。
严格求值 vs 惰性求值
在编程语言理论中，表达式的求值策略分为两类：
| 求值策略 | 定义 | 示例 | 内存模式 |
|---|---|---|---|
| 严格求值 | 表达式一被遇到就计算所有值 | sum([f(x) for x in S]) |
立即分配所有内存 |
| 惰性求值 | 仅在实际需要时才计算值 | Python 生成器 | 按需分配内存 |
生成器是可暂停的状态机
生成器对象保存状态 \(q_t=(p_t,\ell_t)\)：\(p_t\) 是下一条待执行指令的位置，\(\ell_t\) 是局部变量与当前迭代器状态。每次调用 next() 都触发状态转移
\[ (q_t,\operatorname{next})\longmapsto (y_t,q_{t+1}), \tag{2.6}\]
式 式 2.6 给出本节后续实现与解释采用的数学关系。
即执行到下一个 yield，返回 \(y_t\) 并保存 \(q_{t+1}\)；函数结束后转入终止状态并抛出 StopIteration。因此，惰性不是一串预先存好的 yield 表达式，而是“请求一个值、推进一次状态”的执行协议。
流（Streams）的数学定义
在数据处理中，流（Stream）是按顺序逐项提供的记录序列。它可以有有限终点，也可以持续产生而没有预先已知的终点；“流”强调消费协议，不要求对象必然无限。可把流统一写成
\[ S=(s_t)_{t\in T},\qquad T=\{1,\ldots,n\}\ \text{或}\ T=\mathbb N. \]
生成器的惰性求值使有限大文件和潜在无限流都能逐项处理，而不必先物化全部记录。能否保持有限内存还取决于下游操作是否缓存历史状态：
递归、尾递归、记忆化与迭代状态机的区别
递归是函数直接或间接调用自身；尾递归只是“递归调用为最后一步”的一种形式。Python 语言不保证尾调用消除，所以把普通递归改写成尾递归，通常仍会保留 \(O(n)\) 调用栈。记忆化则缓存已经求出的子问题：以朴素递归 Fibonacci 为例，它可把指数级重复计算降为 \(O(n)\) 时间，但缓存和递归栈一般仍需 \(O(n)\) 空间。三者不能互相替代。
下面的 fib_stream 不使用递归，也不使用记忆化；它是一个迭代生成器状态机。状态只有相邻两项 \((a_t,b_t)\)，转移为 \((a_{t+1},b_{t+1})=(b_t,a_t+b_t)\)。生成前 \(n\) 项需要 \(O(n)\) 时间和 \(O(1)\) 辅助状态；若调用者把结果全部收集到列表，列表本身仍会占 \(O(n)\) 空间。
这类状态机适合逐条处理金融数据流：
# 迭代状态机逐项产生斐波那契数，不依赖递归栈或结果缓存
def fib_stream(value_count): # 用显式项数限定本次流消费边界
previous_value, current_value = 0, 1 # 仅保留递推所需的两个相邻状态
for _ in range(value_count): # 每次循环对应一次状态转移
yield current_value # 产出当前值，并把局部状态暂停在此处
previous_value, current_value = current_value, previous_value + current_value # 同步推进下一对状态数据管道（Pipeline）的数学模型
生成器使数据管道成为可能：
\[ \mathrm{数据} \rightarrow \mathrm{过滤} \rightarrow \mathrm{转换} \rightarrow \mathrm{聚合} \]
每个阶段都是一个惰性操作：
这种函数式编程风格在量化金融中非常重要，特别是处理大规模 tick 级数据时。
关键概念: 迭代器与生成器
对于处理大规模经济或金融数据集（如逐笔成交数据），理解生成器的“惰性求值 (lazy evaluation)”特性至关重要。
yield 的生成器函数或生成器表达式创建。它按需产出值，但暂停时仍会保留执行帧、局部变量、当前子迭代器以及这些对象引用的内存；因此其空间开销取决于所保留的状态，不能概括为“只保存一个元素”。当生成器的暂停状态保持有界、下游又及时消费结果时，这种机制可以显著降低物化大规模数据流的峰值内存，是量化金融工程中的重要工具。
def squares(n=10): # 定义生成器函数，生成前n个自然数的平方
print(f'生成从1到{n}的平方数') # 打印提示信息（调用时才执行）
for i in range(1, n + 1): # 从1迭代到n
yield i ** 2 # yield产出平方值，函数在此暂停
# 当你调用生成器时，没有代码会立即执行
gen = squares() # 创建生成器对象（此时函数体未执行）
print(gen) # 打印生成器对象信息，如<generator object squares at 0x...>
# 直到你请求元素时，它才开始执行
for x in gen: # 迭代生成器，触发函数从上次yield处继续执行
print(x, end=' ') # 输出每个平方数，空格分隔：1 4 9 16 25 36 49 64 81 100<generator object squares at 0x74a8bdc67290>
生成从1到10的平方数
1 4 9 16 25 36 49 64 81 100 制作生成器的另一种方法是使用生成器表达式 (generator expression)。这是列表推导式的生成器版本。要创建一个，将本应是列表推导式的内容用圆括号括起来。
<generator object <genexpr> at 0x74a8bdc67300>这等同于更冗长的生成器：
生成器表达式可以用作函数参数，代替列表推导式，这样可以更节省内存，也可能更快。
标准库 itertools 模块提供许多返回迭代器的数据算法。例如，groupby 接受一个可迭代对象和一个键函数，只把相邻且键相同的元素归为一组；若要得到全局按键分组的结果，必须先用同一个键函数排序。
import itertools # 导入迭代器工具模块
names = ['Alan', 'Adam', 'Wes', 'Will', 'Albert', 'Steven'] # 定义列表names
def first_letter(x): # 把姓名首字母定义为分组键
return x[0] # 返回首字符，而不是整个姓名
names_by_letter = sorted(names, key=first_letter) # 先排序，使同首字母姓名相邻
for letter, names_group in itertools.groupby(names_by_letter, first_letter): # 聚合同一首字母的相邻游程
print(letter, list(names_group)) # 子组是一次性迭代器，物化后查看成员A ['Alan', 'Adam', 'Albert']
S ['Steven']
W ['Wes', 'Will']表 2.3 列出了一些其他有用的 itertools 函数。
| 函数 | 描述 |
|---|---|
chain(*iterables) |
通过将迭代器链接在一起来生成一个序列。 |
combinations(iterable, k) |
生成所有可能的k元组，忽略顺序且无放回。 |
permutations(iterable, k) |
生成所有可能的k元组，考虑顺序。 |
groupby(iterable[, keyfunc]) |
为相邻的同键游程生成（键，子迭代器）；全局按键分组前须先按同一键排序。 |
product(*iterables, repeat=1) |
生成输入可迭代对象的笛卡尔积（作为元组）。 |
当我们处理真实世界的金融数据时，数据质量问题几乎是不可避免的——缺失值、格式不符、异常字符等层出不穷。如果程序遇到错误就崩溃，将无法完成对大批量数据的处理。因此，错误与异常处理 (Exception Handling) 是构建健壮、可靠的数据分析程序的必备技能。
Python 中的异常是一种表示程序运行时错误的对象。当错误发生时，程序会”抛出”（raise）一个异常；如果没有代码捕获它，程序就会终止并打印错误信息。例如，float() 在输入不当时会失败，并引发 ValueError：
--------------------------------------------------------------------------- ValueError Traceback (most recent call last) Cell In[95], line 1 ----> 1 float('something') # 尝试将非数字字符串转为浮点数，引发ValueError ValueError: could not convert string to float: 'something'
假设我们想要一个能够优雅失败的 float 版本。我们可以通过编写一个函数，将对 float 的调用封装在 try/except 块中来实现。
1.2345
something你可能只想抑制 ValueError，因为 TypeError 可能表明你的程序中存在一个真正的 bug。要做到这一点，在 except 后面写上异常类型：
--------------------------------------------------------------------------- TypeError Traceback (most recent call last) Cell In[97], line 8 5 return x # ValueError时返回原始值 7 # 这现在会引发 TypeError，正如预期的那样 ----> 8 attempt_float_specific((1, 2)) # 传入元组，触发TypeError（未被捕获，程序报错） Cell In[97], line 3, in attempt_float_specific(x) 1 def attempt_float_specific(x): # 定义只捕获特定异常的转换函数 2 try: # 尝试执行转换 ----> 3 return float(x) # 尝试将x转为浮点数 4 except ValueError: # 只捕获ValueError，其他异常会继续抛出 5 return x TypeError: float() argument must be a string or a real number, not 'tuple'
你可以通过编写一个异常类型的元组来捕获多种异常类型：
在某些情况下，你可能希望无论 try 块是否成功，都执行一些代码。为此，使用 finally。类似地，你可以使用 else 来执行仅在 try 块成功时才执行的代码。
# 虚构的文件写入示例
file_handle = open(path, mode='w', encoding='utf-8') # 用显式编码打开题设路径；展示块不会实际执行
try: # 把可能失败的写入操作限定在异常处理范围内
write_to_file(file_handle) # 将待写内容交给题设中的写入函数
except OSError: # 只处理文件系统相关异常，不吞掉编程错误
print('失败') # 向调用者报告本次文件写入未完成
else: # 仅在写入没有抛出异常时进入成功分支
print('成功') # 确认数据已按预期写入
finally: # 无论成功或失败都释放底层文件描述符
file_handle.close() # 关闭资源，避免句柄泄漏本书大部分内容使用像 pandas.read_csv 这样的高级工具来读取数据文件。然而，理解如何在 Python 中处理文件的基础知识很重要。要打开一个文件进行读写，使用内置的 open 函数。一个最佳实践是传递一个 encoding 参数。
默认情况下，文件以只读模式（'r'）打开。然后我们可以像处理列表一样处理文件对象 f，并迭代其行。行内容会带有行尾（EOL）标记，所以你通常会用 rstrip() 来移除它们。
当你使用 open 创建文件对象时，建议在完成后关闭文件。with 语句通过在退出块时自动关闭文件，使这变得更容易。
表 2.4 列出了有效的文件读/写模式。
| 模式 | 描述 |
|---|---|
r |
只读模式 |
w |
只写模式；创建一个新文件（擦除现有数据） |
x |
只写模式；创建一个新文件，但如果路径已存在则失败 |
a |
追加到现有文件（如果文件不存在则创建） |
r+ |
读写 |
b |
添加到模式中用于二进制文件（例如，'rb' 或 'wb'） |
t |
文本模式文件（默认）；自动将字节解码为 Unicode |
对于可读文件，一些最常用的方法是 read、seek 和 tell。read 从文件中返回一定数量的字符。
# 假设 path 指向一个 utf-8 编码的文件
with open(path, encoding='utf-8') as f:
# 读取前10个字符
content_text = f.read(10)
# 获取当前位置
position_text = f.tell()
# 将位置移动到第3个字节
f.seek(3)
# 从新位置读取一个字符
char_after_seek = f.read(1)
with open(path, mode='rb') as f: # 二进制模式
# 读取前10个字节
content_binary = f.read(10)
# 获取当前位置
position_binary = f.tell()
# 用于演示的虚构输出
print(f'文本内容: "{content_text}", 位置: {position_text}')
print(f'二进制内容: {content_binary}, 位置: {position_binary}') # 展示按字节读取后的内容与文件指针
print(f'移动后读取的字符: "{char_after_seek}"') # 展示seek改变位置后读取的单个字符表 2.5 总结了许多最常用的文件方法。
| 方法/属性 | 描述 |
|---|---|
read([size]) |
以字符串或字节形式返回文件数据，可选大小。 |
readlines([size]) |
返回文件中的行列表，可选大小参数。 |
write(string) |
将传递的字符串写入文件。 |
writelines(strings) |
将传递的字符串序列写入文件。 |
close() |
关闭文件对象。 |
flush() |
将内部 I/O 缓冲区刷新到磁盘。 |
seek(pos) |
移动到指定的文件位置（整数）。 |
tell() |
以整数形式返回当前文件位置。 |
closed |
如果文件已关闭，则为 True。 |
encoding |
用于将字节解释为 Unicode 的编码。 |
请先完成同一难度层级中的题面，再展开完整解答。网页中的解答默认折叠；浏览器打印或导出 PDF 时，下面的打印逻辑会临时展开全部解答，完成后恢复原有开合状态。
习题 2.1: 列表与字符串操作
给定一个字符串列表：
portfolio_copy = portfolio[:] 后修改嵌套持股数量是否会影响原列表，并用原地修改验证习题 2.1 完整解答
代码清单 列表 2.2 给出本题的可复核解答。
stocks = ['恒瑞医药', '宁波港', '宁波银行', '上汽集团', '杭州银行'] # 用五家长三角上市公司构造名称列表
codes = ['600276.SH', '601018.SH', '002142.SZ', '600104.SH', '600926.SH'] # 按相同顺序给出对应证券代码
# (a) 配对股票名称和代码
paired = list(zip(stocks, codes)) # 使用zip配对两个列表并转为列表
print('(a) 股票配对:') # 标明下方逐行输出五家公司名称与同位置证券代码的配对结果
for stock, code in paired: # 逐一解包五个(name, code)元组，核对zip保持输入顺序
print(f'{stock}: {code}') # 打印股票名称与其对应代码
# (b) 创建字典列表
stock_list = [{'name': s, 'code': c} for s, c in paired] # 列表推导式：将配对转为字典列表
print('\n(b) 字典列表:') # 标明输出对象由五个含name/code键的记录字典组成
print(stock_list) # 输出字典列表内容
# (c) 筛选以"宁波"开头的股票
ningbo_stocks = [s for s in stocks if s.startswith('宁波')] # 检查字符串前缀
print(f'\n(c) 宁波相关股票: {ningbo_stocks}') # 输出以'宁波'开头的筛选结果
# (d) 按字母顺序排序
sorted_stocks = sorted(stocks) # 使用sorted按Unicode编码排序股票名称
print(f'\n(d) 排序后: {sorted_stocks}') # 输出按Unicode编码排序后的股票列表(a) 股票配对:
恒瑞医药: 600276.SH
宁波港: 601018.SH
宁波银行: 002142.SZ
上汽集团: 600104.SH
杭州银行: 600926.SH
(b) 字典列表:
[{'name': '恒瑞医药', 'code': '600276.SH'}, {'name': '宁波港', 'code': '601018.SH'}, {'name': '宁波银行', 'code': '002142.SZ'}, {'name': '上汽集团', 'code': '600104.SH'}, {'name': '杭州银行', 'code': '600926.SH'}]
(c) 宁波相关股票: ['宁波港', '宁波银行']
(d) 排序后: ['上汽集团', '宁波港', '宁波银行', '恒瑞医药', '杭州银行']代码清单 列表 2.3 给出本题的可复核解答。
portfolio = [['宁波港', 100], ['恒瑞医药', 200]] # 构造含可变嵌套列表的两项持仓题设
portfolio_copy = portfolio[:] # 浅复制外层列表，内层持仓记录仍共享对象
portfolio_copy[0][1] = 120 # 通过副本原地修改共享的第一项持股数量
assert portfolio[0][1] == 120 # 验证浅复制不会隔离嵌套可变对象的修改
tuple_key = ('600276.XSHG', 2024) # 只含不可变元素的元组可计算稳定哈希值
security_lookup = {tuple_key: '恒瑞医药'} # 用可哈希元组作为证券年度复合键
assert security_lookup[tuple_key] == '恒瑞医药' # 验证元组键可以可靠查找
assert isinstance(hash(tuple_key), int) # 显式验证该元组具有整数哈希值列表和字典是可变对象，不能直接作为字典键；只含可哈希元素的元组可以作为键。若元组内部包含列表，该元组仍不可哈希，因此不能只凭最外层类型判断。
习题 2.2: 字典操作
下面给定一组只用于字典操作的固定数值（标签沿用省份名，数值和汇率均是题设参数，不作为官方统计或实时汇率）：
习题 2.2 完整解答
代码清单 列表 2.4 给出本题的可复核解答。
gdp_2022 = { # 定义长三角地区省份GDP字典（万亿元）
'江苏': 12.29, # 江苏2022年GDP（万亿元）
'浙江': 7.77, # 浙江2022年GDP（万亿元）
'上海': 4.47, # 上海2022年GDP（万亿元）
'安徽': 4.50 # 安徽2022年GDP（万亿元）
} # 字典定义结束
# (a) 添加福建
gdp_2022['福建'] = 5.31 # 向字典添加福建GDP数据
print('(a) 添加福建后的字典:') # 标明输出已在原四省市GDP映射中新增福建及5.31万亿元题设值
print(gdp_2022) # 输出完整字典内容
# (b) 计算总GDP
total_gdp = sum(gdp_2022.values()) # 对字典所有值求和得到总GDP
print(f'\n(b) 总GDP: {total_gdp:.2f} 万亿元') # 输出长三角各省GDP合计
# (c) 找出GDP最高的省份
max_province = max(gdp_2022, key=gdp_2022.get) # 用max+key参数找出GDP最大的省份名
max_gdp = gdp_2022[max_province] # 获取该省份的GDP值
print(f'\n(c) GDP最高省份: {max_province} ({max_gdp} 万亿元)') # 输出GDP最高的省份名称及数值
# (d) 转换为美元
exchange_rate = 7.0 # 1美元=7人民币
gdp_usd = {k: v / exchange_rate for k, v in gdp_2022.items()} # 字典推导式：将每个GDP值转换为美元
print(f'\n(d) GDP（万亿美元）:') # 输出人民币转美元后的标题
for province, gdp in gdp_usd.items(): # 按省份键逐项报告以7元/美元换算后的万亿美元值
print(f' {province}: {gdp:.3f}') # 逐省输出换算后的美元GDP
# (e) 按GDP降序排列
sorted_gdp = dict(sorted(gdp_2022.items(), key=lambda x: x[1], reverse=True)) # 按GDP值降序排序并转回字典
print(f'\n(e) 降序排列:') # 输出GDP降序排列标题
for province, gdp in sorted_gdp.items(): # 按GDP降序后的插入顺序逐省输出人民币万亿元值
print(f' {province}: {gdp} 万亿元') # 逐省输出降序排列后的GDP(a) 添加福建后的字典:
{'江苏': 12.29, '浙江': 7.77, '上海': 4.47, '安徽': 4.5, '福建': 5.31}
(b) 总GDP: 34.34 万亿元
(c) GDP最高省份: 江苏 (12.29 万亿元)
(d) GDP（万亿美元）:
江苏: 1.756
浙江: 1.110
上海: 0.639
安徽: 0.643
福建: 0.759
(e) 降序排列:
江苏: 12.29 万亿元
浙江: 7.77 万亿元
福建: 5.31 万亿元
安徽: 4.5 万亿元
上海: 4.47 万亿元习题 2.3: 集合操作
给定两个集合：
习题 2.3 完整解答
代码清单 列表 2.5 给出本题的可复核解答。
set_a = {'宁波港', '宁波银行', '恒瑞医药', '上汽集团'} # 用四家长三角公司构造证券池A
set_b = {'宁波港', '杭州银行', '南京银行', '上汽集团'} # 用另一组长三角公司构造证券池B
# (a) 交集
intersection = set_a & set_b # 用&运算符计算两个集合的交集
print(f'(a) 交集: {intersection}') # 输出两个集合的共有股票
# (b) 并集
union = set_a | set_b # 用|运算符计算两个集合的并集
print(f'\n(b) 并集: {union}') # 输出两个集合的所有股票（去重）
# (c) 差集（A-B）
difference = set_a - set_b # 用-运算符计算差集（在A中但不在B中）
print(f'\n(c) A-B差集: {difference}') # 输出在A中但不在B中的股票
# (d) 成员测试
is_in_b = '宁波银行' in set_b # 检查宁波银行是否在set_b中，返回布尔值
print(f'\n(d) 宁波银行在set_b中: {is_in_b}') # 输出成员测试的布尔结果(a) 交集: {'上汽集团', '宁波港'}
(b) 并集: {'杭州银行', '宁波银行', '恒瑞医药', '南京银行', '上汽集团', '宁波港'}
(c) A-B差集: {'宁波银行', '恒瑞医药'}
(d) 宁波银行在set_b中: False(e) 数据结构选择：反复做成员检查时应优先使用集合。列表需要从前向后扫描，平均时间复杂度为 \(O(n)\)；集合通过哈希表查找，平均时间复杂度为 \(O(1)\)，但最坏情况仍可能退化，而且集合不保留可用于业务排序的位置语义。若任务还要求稳定顺序，应同时保留列表，或在完成去重/查找后显式排序。
习题 2.4: 函数与递归
定位与产出：核心进阶题；先修为函数、条件分支、循环、生成器小节中的递归与记忆化辨析；预计 45—60 分钟；提交三种实现、前 15 项结果、相邻项比值和时间—空间复杂度比较。
斐波那契数列适合比较递归、循环与记忆化三种实现，因为三者计算同一数学对象，却具有不同的重复计算和内存特征。数列定义为：
习题 2.4 完整解答
(a) 递归版本
代码清单 列表 2.6 给出本题的可复核解答。
(b) 循环版本（更高效）
(c) 记忆化优化版本
from functools import lru_cache # 从functools模块导入lru_cache
@lru_cache(maxsize=None) # 启用无限大小缓存的记忆化装饰器
def fib_memoized(n): # 定义记忆化版斜波那契函数
"""使用记忆化优化递归计算""" # 说明同一n的F(n)会从无界缓存返回，避免重复展开递归树
if n <= 1: # 基本情况：n为0或1时直接返回
return n # 返回F(0)=0或F(1)=1
return fib_memoized(n-1) + fib_memoized(n-2) # 递归计算（已计算过的会直接返回缓存）(d) 计算前15个斐波那契数
(d) 前15个斐波那契数:
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377](e) 计算相邻两项的比值
(e) 相邻项比值（趋近于黄金分割率1.618...）:
F(2)/F(1) = 1.000000
F(3)/F(2) = 2.000000
F(4)/F(3) = 1.500000
F(5)/F(4) = 1.666667
F(6)/F(5) = 1.600000
F(7)/F(6) = 1.625000
F(8)/F(7) = 1.615385
F(9)/F(8) = 1.619048
F(10)/F(9) = 1.617647
F(11)/F(10) = 1.618182
F(12)/F(11) = 1.617978
F(13)/F(12) = 1.618056
F(14)/F(13) = 1.618026
说明：这个比值趋近于黄金分割率 φ ≈ 1.6180339887...斐波那契数列与黄金分割率的数学关系
斐波那契数列相邻两项的比值趋近于黄金分割率 \(\phi\)，这是一个重要的数学常数：
\[ \phi = \frac{1 + \sqrt{5}}{2} \approx 1.6180339887... \]
当 \(n \to \infty\) 时，\(F(n+1)/F(n) \to \phi\)。这是因为斐波那契数列的通项公式（Binet’s Formula）为：
\[ F(n) = \frac{\phi^n - \psi^n}{\sqrt{5}} \]
其中 \(\psi = \frac{1 - \sqrt{5}}{2} \approx -0.618\)。这项练习的评价对象是三种程序实现及其复杂度，不把由黄金分割率派生的比例解释为价格预测依据。
习题 2.5: 文件处理与数据清洗
定位与产出：核心应用题；先修为列表、字典、循环、函数、异常处理、csv 与上下文管理器；预计 60—80 分钟；提交输入合同核验、类型转换、逐日收益、极值记录、输出 CSV 和写回行数检查。
教师会先运行下方脚手架，把本地恒瑞医药真实行情转换为一个临时 CSV（格式：日期,开盘价,最高价,最低价,收盘价,成交量）。该块只用于提供输入并核对真实数据血缘，不属于题目知识点，也不评分。学生只接收 exercise_2_5_input_csv_path 与 exercise_2_5_output_csv_path 两个路径合同变量，使用本章讲过的列表、字典、循环、函数、上下文管理器及下方标准库桥接编写代码：
习题 2.5 完整解答
教师提供的数据准备（只运行核对，不评分）
教师脚手架最后只向学生交付两个路径合同变量：exercise_2_5_input_csv_path 指向已经存在且可读的六列输入 CSV，exercise_2_5_output_csv_path 指向允许学生创建或覆盖的答案输出 CSV。学生答案不得依赖教师块中的临时目录对象、Pandas 对象或任何已导入模块。
代码清单 列表 2.7 给出本题的可复核解答。
from pathlib import Path # 导入跨平台路径类
from tempfile import TemporaryDirectory # 导入临时目录工具
import pandas as pd # 用于从本地HDF5抽取真实输入
exercise_2_5_temporary_directory = TemporaryDirectory() # 使用可自动清理的临时目录，避免练习污染仓库数据
exercise_2_5_input_csv_path = Path(exercise_2_5_temporary_directory.name) / 'stock_data_sample.csv' # 声明学生只读输入路径合同
exercise_2_5_output_csv_path = Path(exercise_2_5_temporary_directory.name) / 'stock_data_cleaned.csv' # 声明学生可写输出路径合同
hengrui_sample_df = ( # 从真实本地行情抽取恒瑞医药的五行教学样本
pd.read_hdf( # 直接从已确认存在的 HDF5 表读取所需公司子集
f'{DATA_ROOT}/stock/stock_price_pre_adjusted.h5', # 复用本章首个代码块定义的跨平台数据根路径
where="order_book_id='600276.XSHG'", # 仅选择江苏恒瑞医药，避免载入全市场
) # 完成选择性读取
.reset_index() # 将日期索引还原为普通列以便写入 CSV
.loc[lambda stock_rows: pd.to_datetime(stock_rows['date']).dt.year.eq(2023)] # 只保留2023年真实观测
.head(5) # 控制练习输入规模，便于逐行核对
.rename(columns={ # 把底表字段映射为题目约定的中文列名
'date': '日期', 'open': '开盘价', 'high': '最高价', # 映射日期与前三个价格字段
'low': '最低价', 'close': '收盘价', 'volume': '成交量', # 映射其余行情字段
}) # 完成列名映射
) # 完成恒瑞医药教学样本构造
hengrui_sample_df[['日期', '开盘价', '最高价', '最低价', '收盘价', '成交量']].to_csv( # 仅导出题目要求的六列
exercise_2_5_input_csv_path, index=False, encoding='utf-8', date_format='%Y-%m-%d' # 固定编码与日期格式以便标准库解析
) # 完成真实行情 CSV 教学副本写出
print(f'教师输入已就绪：{exercise_2_5_input_csv_path.name}，共 {len(hengrui_sample_df)} 行') # 只核对输入文件身份和记录数，不评价Pandas操作教师输入已就绪：stock_data_sample.csv，共 5 行以下才是学生作答与评分范围，全部使用本章已经讲授的 Python 内置能力与标准库。
标准库桥接：CSV 表头、日期格式与路径合同
csv.DictReader 默认把输入首行解释为 fieldnames，随后迭代只返回数据行，因此它完成的是“读取并核对标题”，而不是把标题静默当作普通记录。datetime.strptime(text, format) 会按格式指令核验年月日与日历语义，不合法日期会抛出 ValueError；但 %m 和 %d 并不强制两位零填充。若本题还要求固定词法 YYYY-MM-DD，解析后须用 strftime('%Y-%m-%d') 回写并与原文本比较。写出时，csv.DictWriter 的 fieldnames 同时规定列集合与顺序，必须先调用 writeheader()，再用 writerows() 写入字典记录。
Path.open(...) 是绑定到某个 Path 对象的便捷写法；内置 open(path, ...) 则同时接受字符串和实现 os.PathLike 合同的 Path。两者都返回需要由 with 管理的文件对象。本题读取时演示 Path.open，写出时演示内置 open，不应把两种写法误解为不同文件格式。
import csv # 导入标准库CSV读写器以执行表头和字段顺序合同
from datetime import datetime # 导入日期解析器以核验来源日期的日历语义
from pathlib import Path # 导入实现os.PathLike合同的跨平台路径类
from tempfile import TemporaryDirectory # 导入临时目录工具以隔离桥接示例的文件往返
with TemporaryDirectory() as bridge_temporary_directory: # 将接口演示限制在可自动清理的临时目录
bridge_csv_path = Path(bridge_temporary_directory) / 'bridge.csv' # 构造实现os.PathLike合同的临时Path
with bridge_csv_path.open('w', encoding='utf-8', newline='') as bridge_output_file: # 用Path.open写入Path绑定的文件
bridge_writer = csv.DictWriter(bridge_output_file, fieldnames=['日期', '收盘价']) # 用fieldnames固定输出列集合与顺序
bridge_writer.writeheader() # 先写字段名，保证下游DictReader能识别标题
bridge_writer.writerows([{'日期': '2023-01-03', '收盘价': '38.50'}]) # 再批量写入符合字段合同的字典记录
with open(bridge_csv_path, 'r', encoding='utf-8', newline='') as bridge_input_file: # 用内置open读取同一个PathLike对象
bridge_reader = csv.DictReader(bridge_input_file) # 让首行成为fieldnames并从第二行开始返回字典记录
assert bridge_reader.fieldnames == ['日期', '收盘价'] # 核验来源表头的名称与顺序均符合预期
bridge_source_row = next(bridge_reader) # 取得首条数据而不是再次读到标题行
bridge_date_text = bridge_source_row['日期'] # 保留原始词法，供固定宽度合同核验
bridge_trade_date = datetime.strptime(bridge_date_text, '%Y-%m-%d') # 按题目声明的年月日格式解析日期
assert bridge_trade_date.strftime('%Y-%m-%d') == bridge_date_text # 拒绝非YYYY-MM-DD词法，即使strptime可解析(a) -(c): 读取并解析文件
import csv # 学生答案自行导入CSV标准库，不继承教师脚手架命名空间
from datetime import datetime # 学生答案自行导入日期解析器
from pathlib import Path # 学生答案自行导入路径类以消费输入输出合同
student_input_csv_path = Path(exercise_2_5_input_csv_path) # 只从显式合同取得可读输入路径
expected_input_fieldnames = ['日期', '开盘价', '最高价', '最低价', '收盘价', '成交量'] # 固定来源表头名称与顺序
stock_price_records = [] # 收集已通过字段和类型检查的逐日行情记录
with student_input_csv_path.open('r', encoding='utf-8', newline='') as csv_input_file: # 用Path.open读取合同文件并确保自动关闭
stock_price_reader = csv.DictReader(csv_input_file) # 把首行解释为字段名并从第二行开始返回字典
assert stock_price_reader.fieldnames == expected_input_fieldnames # 在解析数值前阻断缺列、错名或顺序漂移
for source_price_row in stock_price_reader: # 逐行处理表头之后的真实行情记录
# 转换数据类型
source_date_text = source_price_row['日期'] # 保留CSV中的原始日期词法以核验固定宽度合同
parsed_trade_date = datetime.strptime(source_date_text, '%Y-%m-%d') # 先核验年月日与日历语义
assert parsed_trade_date.strftime('%Y-%m-%d') == source_date_text # 再拒绝可解析但未零填充的日期文本
parsed_price_record = { # 构建字段名稳定且类型明确的单条股价记录
'date': parsed_trade_date, # 保存同时通过日历语义和固定词法核验的交易日
'open': float(source_price_row['开盘价']), # 将开盘价转换为可计算浮点数
'high': float(source_price_row['最高价']), # 将最高价转换为可计算浮点数
'low': float(source_price_row['最低价']), # 将最低价转换为可计算浮点数
'close': float(source_price_row['收盘价']), # 将收盘价转换为可计算浮点数
'volume': int(float(source_price_row['成交量'])) # 兼容CSV中整数或带小数点文本形式的成交量
} # 记录字典定义结束
stock_price_records.append(parsed_price_record) # 保存已完成类型转换的记录供后续评分步骤复用
print('(a)-(c) 读取的数据:') # 标明输出是CSV解析后按日期展示的收盘价字段，不含尚未计算的涨跌幅
for price_record in stock_price_records: # 遍历每条已通过表头和类型合同的股价记录
print(f" {price_record['date'].strftime('%Y-%m-%d')}: 收盘价={price_record['close']}") # 打印日期和收盘价(a)-(c) 读取的数据:
2023-01-03: 收盘价=37.9796
2023-01-04: 收盘价=38.3056
2023-01-05: 收盘价=39.0763
2023-01-06: 收盘价=38.8392
2023-01-09: 收盘价=39.0763(d) 计算每日涨跌幅
for record_index in range(len(stock_price_records)): # 按日期顺序遍历每个已解析交易日
if record_index == 0: # 首个交易日无样本内前收盘价
stock_price_records[record_index]['change_pct'] = 0.0 # 以题目约定的0占位标记首日不可计算涨跌幅
else: # 非首日
previous_close_price = stock_price_records[record_index - 1]['close'] # 取得同一排序样本的前一日收盘价
current_close_price = stock_price_records[record_index]['close'] # 取得当前交易日收盘价
stock_price_records[record_index]['change_pct'] = (current_close_price - previous_close_price) / previous_close_price * 100 # 按相邻收盘价计算简单收益百分比
print('\n(d) 添加涨跌幅后的数据:') # 输出计算涨跌幅后的标题
for price_record in stock_price_records: # 遍历已经补入涨跌幅字段的全部记录
if price_record['change_pct'] != 0: # 按题目约定跳过首日0占位记录
print(f" {price_record['date'].strftime('%Y-%m-%d')}: {price_record['change_pct']:+.2f}%") # 输出可计算日期的涨跌幅百分比
(d) 添加涨跌幅后的数据:
2023-01-04: +0.86%
2023-01-05: +2.01%
2023-01-06: -0.61%
2023-01-09: +0.61%(e) 找出收益率最大和最小的交易日
maximum_return_record = max(stock_price_records[1:], key=lambda price_record: price_record['change_pct']) # 在可计算区间中找出收益率最大日
minimum_return_record = min(stock_price_records[1:], key=lambda price_record: price_record['change_pct']) # 在可计算区间中找出收益率最小日
print(f'\n(e) 收益率最大: {maximum_return_record["date"].strftime("%Y-%m-%d")} ({maximum_return_record["change_pct"]:+.2f}%)') # 输出最大收益日及带符号收益
print(f' 收益率最小: {minimum_return_record["date"].strftime("%Y-%m-%d")} ({minimum_return_record["change_pct"]:+.2f}%)') # 输出最小收益日而不预设其一定为负
(e) 收益率最大: 2023-01-05 (+2.01%)
收益率最小: 2023-01-06 (-0.61%)(f) 写入新文件
student_output_csv_path = Path(exercise_2_5_output_csv_path) # 在实际写出阶段从显式合同取得可写路径
output_fieldnames = ['date', 'open', 'high', 'low', 'close', 'volume', 'change_pct'] # 固定答案CSV的字段集合与顺序
cleaned_output_records = [{**price_record, 'date': price_record['date'].strftime('%Y-%m-%d')} for price_record in stock_price_records] # 将日期恢复为合同规定的年月日文本
with open(student_output_csv_path, 'w', encoding='utf-8', newline='') as csv_output_file: # 演示内置open同样接受Path合同并自动关闭文件
stock_price_writer = csv.DictWriter(csv_output_file, fieldnames=output_fieldnames) # 让DictWriter按声明顺序序列化字典字段
stock_price_writer.writeheader() # 先写标题以保证输出可被DictReader直接回读
stock_price_writer.writerows(cleaned_output_records) # 一次写入全部符合字段合同的清洗记录
print(f'\n(f) 已将清洗后的数据写入合同文件: {student_output_csv_path.name}') # 报告显式输出合同对应的文件名
(f) 已将清洗后的数据写入合同文件: stock_data_cleaned.csv本题所需标准库桥接：math.log 与 math.sqrt
对数收益需要自然对数，波动率需要平方根。二者都来自 Python 标准库 math，不是 NumPy API。若简单收益列表为 \(r_1,\ldots,r_n\)，其样本均值与样本标准差按下式计算：
\[ \bar r=\frac{1}{n}\sum_{i=1}^{n}r_i,\qquad s=\sqrt{\frac{\sum_{i=1}^{n}(r_i-\bar r)^2}{n-1}}. \]
习题 2.6: 股票收益率计算
定位与产出：核心应用题；先修为列表、字典、相邻记录遍历、函数以及 math 标准库；预计 45—60 分钟；提交输入顺序断言、两类收益、正负收益计数、样本波动率和最大涨跌记录。
教师会先运行下方脚手架，把本地宁波港（601018）最近 100 个真实交易日转换为按日期升序排列的 ningbo_price_records；每个元素只含 date 字符串和 close 浮点数。该脚手架中的 Pandas/HDF5 操作只负责保持真实数据链，学生只运行核对，不解释、不修改、不评分。学生从这份“日期—价格”字典列表开始，使用循环、函数以及 math、statistics 标准库编写代码：
习题 2.6 完整解答
教师提供的数据准备（只运行核对，不评分）
代码清单 列表 2.8 给出本题的可复核解答。
import pandas as pd # 由教师使用Pandas把真实HDF5行情转换为本章可处理的内置列表
ningbo_port_source_df = pd.read_hdf( # 只读取宁波港真实前复权行情以建立输入血缘
f'{DATA_ROOT}/stock/stock_price_pre_adjusted.h5', # 复用首块DATA_ROOT定位A股前复权HDF5文件
where="order_book_id='601018.XSHG'" # 下推单证券条件，避免载入全市场记录
).reset_index() # 把交易日索引恢复为列，以便教师按日期整理
ningbo_port_source_df = ningbo_port_source_df.sort_values('date').tail(100) # 固定取按日期排序后的最近100个真实观测
ningbo_price_records = [ # 转为学生已学的字典列表，隔离尚未讲授的DataFrame接口
{'date': trade_date.strftime('%Y-%m-%d'), 'close': float(close_price)} # 每条记录只保留ISO日期和正收盘价
for trade_date, close_price in ningbo_port_source_df[['date', 'close']].itertuples(index=False, name=None) # 保持真实行情行序
] # 完成教师提供的标准库友好输入
print(f'教师输入已就绪：{len(ningbo_price_records)} 行，日期 {ningbo_price_records[0]["date"]} 至 {ningbo_price_records[-1]["date"]}') # 只核对输入范围教师输入已就绪：100 行，日期 2025-08-06 至 2025-12-31以下才是学生作答与评分范围，不使用 Pandas 或 NumPy。
(a) 核对输入合同
assert len(ningbo_price_records) == 100 # 核对教师输入保持题目约定的100个交易日
assert all(record['close'] > 0 for record in ningbo_price_records) # 对数变换前排除非正价格
record_dates = [record['date'] for record in ningbo_price_records] # 提取ISO日期以检查输入行序
assert record_dates == sorted(record_dates) # ISO日期按字典序应与交易日升序完全一致
print(f'(a) 输入合同通过：{record_dates[0]} 至 {record_dates[-1]}') # 报告已核对的真实样本范围(a) 输入合同通过：2025-08-06 至 2025-12-31(b) 计算收益率
def calculate_return_records(price_records): # 把日期—价格列表转换为逐期收益记录
return_records = [] # 首日没有前值，因此结果从第二个交易日开始
for previous_record, current_record in zip(price_records, price_records[1:]): # 成对遍历相邻真实交易日
price_ratio = current_record['close'] / previous_record['close'] # 计算当期末价相对前期末价的倍数
return_records.append({ # 保存日期与两种同区间收益，供后续统计复用
'date': current_record['date'], # 收益归属于当前交易日
'simple_return': price_ratio - 1.0, # 价格倍数减1得到简单收益
'log_return': log(price_ratio), # 价格倍数取自然对数得到对数收益
}) # 完成一条收益记录
return return_records # 返回99个相邻交易区间的收益字典
ningbo_return_records = calculate_return_records(ningbo_price_records) # 调用本章函数完成真实价格转换
print('(b) 最近5个收益区间:') # 标明下方记录只使用内置字典与列表
print(ningbo_return_records[-5:]) # 用负切片展示末五个真实收益区间(b) 最近5个收益区间:
[{'date': '2025-12-25', 'simple_return': 0.005449591280654031, 'log_return': 0.005434795985956995}, {'date': '2025-12-26', 'simple_return': -0.005420054200542035, 'log_return': -0.005434795985956945}, {'date': '2025-12-29', 'simple_return': -0.00544959128065392, 'log_return': -0.0054644944720787375}, {'date': '2025-12-30', 'simple_return': -0.005479452054794498, 'log_return': -0.005494519317640702}, {'date': '2025-12-31', 'simple_return': 0.0, 'log_return': 0.0}](c) 统计正负收益天数
positive_days = sum(record['simple_return'] > 0 for record in ningbo_return_records) # 布尔值按1/0求和得到上涨区间数
negative_days = sum(record['simple_return'] < 0 for record in ningbo_return_records) # 同口径统计下跌区间数
unchanged_days = len(ningbo_return_records) - positive_days - negative_days # 由总区间数扣除涨跌得到平盘数
print(f'\n(c) 上涨 {positive_days} 天，下跌 {negative_days} 天，平盘 {unchanged_days} 天') # 完整报告三类互斥结果
(c) 上涨 43 天，下跌 42 天，平盘 14 天(d) 计算统计量
simple_returns = [record['simple_return'] for record in ningbo_return_records] # 提取同口径简单收益序列
mean_return = sum(simple_returns) / len(simple_returns) # 按正文公式计算真实收益的算术平均值
squared_deviations = [(simple_return - mean_return) ** 2 for simple_return in simple_returns] # 计算每期收益对均值的平方偏差
daily_volatility = sqrt(sum(squared_deviations) / (len(simple_returns) - 1)) # 按n-1分母和正文公式计算样本日波动率
annualized_volatility = daily_volatility * sqrt(252) # 按一年252个交易日的平方根规则年化
print(f'\n(d) 日均收益率: {mean_return:.4%}') # 以百分比展示日均简单收益
print(f' 样本日波动率: {daily_volatility:.4%}') # 明确标准差采用样本口径
print(f' 年化波动率: {annualized_volatility:.2%}') # 展示252日年化结果
(d) 日均收益率: 0.0099%
样本日波动率: 0.9855%
年化波动率: 15.64%(e) 最大涨跌幅
maximum_gain_record = max(ningbo_return_records, key=lambda record: record['simple_return']) # 按简单收益键找出最大涨幅记录
maximum_loss_record = min(ningbo_return_records, key=lambda record: record['simple_return']) # 按相同键找出最大跌幅记录
print(f'\n(e) 最大涨幅: {maximum_gain_record["date"]}，{maximum_gain_record["simple_return"]:+.2%}') # 报告最大涨幅日期与幅度
print(f' 最大跌幅: {maximum_loss_record["date"]}，{maximum_loss_record["simple_return"]:+.2%}') # 报告最大跌幅日期与幅度
(e) 最大涨幅: 2025-09-23，+3.24%
最大跌幅: 2025-09-25，-2.95%简单收益率与对数收益率的对比
在金融计量学中，理解两类收益率的区别至关重要：
对数收益率的优势： - 时间可加性: \(T\) 期对数收益率等于各单期对数收益率之和，即 \(r_{0,T} = \sum_{t=1}^{T} r_t\)。 - 正确理解涨跌复合: 上涨 10% 后再下跌 10% 时，对数收益率之和为 \(\log(1.1)+\log(0.9)=\log(0.99)\approx-0.01005\)，并不等于 0；终值是初值的 \(0.99\)，即累计简单收益为 \(-1\%\)。若要求终值回到原点，从上涨 10% 后的价格出发应下跌约 \(9.09\%\)。正负等幅的对数收益相加才为 0。 - 条件性正态性: 在几何布朗运动等连续时间模型中，固定时间区间的对数收益被假设为正态分布；这不是所有真实市场收益或任意持有期都自动满足的经验事实。
简单收益率直接对应财富变化比例；对数收益率具有跨期可加性。选择哪一种取决于分析任务，不能仅按“展示”或“研究”给出无条件优先级。
习题 2.7: 生成器与内存效率
定位与产出：核心综合题；先修为生成器表达式、生成器函数、局部随机数状态和峰值内存口径；预计 40—55 分钟；提交三种实现、同序列校验和、tracemalloc 峰值比较及适用边界说明。
在处理大量金融数据时，内存效率很重要。比较三种方法：
习题 2.7 完整解答
代码清单 列表 2.9 给出本题的可复核解答。
(a) 列表方法
(b) 生成器表达式
(c) 生成器函数
(d) 比较内存占用
def measure_peak_and_sum(factory, n, seed): # 对“创建并完整消费”实施统一测量
gc.collect() # 降低上一轮对象对峰值的干扰
tracemalloc.start() # 从本方法开始跟踪Python内存分配
values = factory(n, seed) # 列表立即分配，生成器只建立暂停状态
checksum = sum(values) # 三种方法都完成相同的全量聚合任务
_, peak_bytes = tracemalloc.get_traced_memory() # 读取从创建到消费结束的峰值
tracemalloc.stop() # 结束本轮，避免不同方法相互污染
return peak_bytes, checksum # 同时返回资源指标与结果证据
factories = { # 固定三种待比较的实现
'列表': get_random_list,
'生成器表达式': get_random_generator_expr,
'生成器函数': get_random_generator_func,
}
memory_results = { # 对每种实现独立运行完整计算
name: measure_peak_and_sum(factory, n, experiment_seed)
for name, factory in factories.items()
}三种实现都消费完整序列后，再比较数值校验和与峰值内存：
reference_sum = memory_results['列表'][1] # 以列表结果作为数值一致性基准
assert all( # 三种方法必须消费完全相同的随机序列
math.isclose(result_sum, reference_sum, rel_tol=0.0, abs_tol=1e-12)
for _, result_sum in memory_results.values()
)
assert memory_results['列表'][0] > memory_results['生成器表达式'][0] # 流式表达式应降低峰值
assert memory_results['列表'][0] > memory_results['生成器函数'][0] # 流式函数也应降低峰值
print(f'(d) 完整消费同一随机序列的峰值内存 (n={n:,}):') # 报告可比实验口径
# 逐项报告三种容器策略的峰值字节数与同口径聚合结果
for method_name, (peak_bytes, result_sum) in memory_results.items():
print(f' {method_name}: {peak_bytes:,} 字节；总和={result_sum:.6f}') # 同列资源与结果证据
print('断言通过：三种方法的聚合结果一致，且本次运行中两种生成器的峰值低于列表。') # 汇总数值一致性与峰值比较的断言结果(d) 完整消费同一随机序列的峰值内存 (n=100,000):
列表: 3,205,601 字节；总和=49974.745038
生成器表达式: 4,040 字节；总和=49974.745038
生成器函数: 3,628 字节；总和=49974.745038
断言通过：三种方法的聚合结果一致，且本次运行中两种生成器的峰值低于列表。这里比较的是 tracemalloc 观测到的 Python 分配峰值，覆盖对象创建与完整消费过程，而不是 sys.getsizeof(list) 只统计列表容器本身的浅层大小。具体字节数会随 Python 版本和运行环境变化；结论也不意味着生成器“没有内存成本”，因为生成器仍保留暂停帧、局部随机数生成器及其引用。对这个逐项求和任务，列表需要同时保存全部浮点数，而两种生成器无需物化完整序列，所以峰值通常明显较低。
生成器在金融大数据中的应用
在处理 A 股高频交易（Tick-by-Tick）数据时，数据量往往以 GB 甚至 TB 计。
核心应用模式： - 流式处理: 逐行读取超大型 CSV 或 H5 文件，避免一次性加载导致的内存崩溃（OOM）。 - 数据管道 (Pipeline): 通过链接多个生成器（如：读取 -> 清洗 -> 聚合），每个环节只处理当前记录。
这种模式适合逐批读取无法一次装入内存的记录。是否真正降低端到端内存和运行时间，还取决于文件格式、解码缓冲、下游聚合状态与是否最终物化全部结果。
本章建立的是容器、函数、生成器和文件处理基础。完整回测还需要类与实例、dataclass、类型标注、Pandas 时间索引、复权口径、信息时点、下一可交易日执行、手续费、印花税、滑点、100 股交易单位、停牌与涨跌停等知识；这些条件将在后续数据处理和案例章节逐步建立。
因此，本章不提供策略收益或买卖判断。学完后续章节再回到这一问题时，可以把股票数据、组合账本和执行规则分别封装为对象，并用守恒断言检查现金、持仓和成交时点。均线的相对位置只能定义一个可计算状态；没有时间外检验和市场摩擦模型时，它不构成预测证据或投资行动建议。
本章介绍了 Python 的核心内置数据结构——元组、列表、字典和集合，以及函数、生成器与文件操作。ADT 描述接口行为，具体数据结构说明实现；列表保留顺序和重复元素，流既可以有限也可以没有预知终点。通过 CSV 的上下文管理、字段类型转换、写回与记录数核对，读者已经完成一次可审计的文件往返；通过教师提供的真实行情列表，也已用本章范围内的循环、函数和标准库计算简单收益率与波动率，并把真实观测与题设常量分开。掌握这些语义、可哈希性与复杂度边界，为后续使用 NumPy 和 Pandas 处理数据建立了可靠基础。
下一章转向 NumPy 的数组、广播和向量化计算，把本章逐个对象的操作扩展为具有明确形状和轴语义的批量计算。